@Test public void input_hedaer() throws Exception { MockFormat format = format(3, HeaderType.FORCE); String[][] data = { { "A", "B", "C", }, { "D", "E", "F", }, { "G", "H", "I", }, }; try (ModelInput<String[]> in = format.createInput(String[].class, "dummy", input(data))) { String[][] result = collect(3, in); assertThat(result, is(Arrays.copyOfRange(data, 1, 3))); } }
@Override public ModelInput<T> createInput( Class<? extends T> dataType, String path, InputStream stream, long offset, long fragmentSize) throws IOException, InterruptedException { TextFormat format = getTextFormat(); RecordDefinition<T> record = getRecordDefinition(); Set<InputOption> options = offset == 0 ? INPUT_OPTS_HEAD : INPUT_OPTS_REST; InputStream source = decorate(stream, offset, fragmentSize); return new DecoratedInput(record.newInput(format.open(source), path, options), new StringOption(path)); }
AbstractTextStreamFormat extends ConfigurableBinaryStreamFormat<T> { @Override public ModelInput<T> createInput( Class<? extends T> dataType, String path, InputStream stream, long offset, long fragmentSize) throws IOException, InterruptedException { TextFormat format = getTextFormat(); RecordDefinition<T> record = getRecordDefinition(); Set<InputOption> options = offset == 0 ? INPUT_OPTS_HEAD : INPUT_OPTS_REST; InputStream source = decorate(stream, offset, fragmentSize); return new DecoratedInput(record.newInput(format.open(source), path, options), new StringOption(path)); } }
AbstractTextStreamFormat extends ConfigurableBinaryStreamFormat<T> { @Override public ModelInput<T> createInput( Class<? extends T> dataType, String path, InputStream stream, long offset, long fragmentSize) throws IOException, InterruptedException { TextFormat format = getTextFormat(); RecordDefinition<T> record = getRecordDefinition(); Set<InputOption> options = offset == 0 ? INPUT_OPTS_HEAD : INPUT_OPTS_REST; InputStream source = decorate(stream, offset, fragmentSize); return new DecoratedInput(record.newInput(format.open(source), path, options), new StringOption(path)); }  }
AbstractTextStreamFormat extends ConfigurableBinaryStreamFormat<T> { @Override public ModelInput<T> createInput( Class<? extends T> dataType, String path, InputStream stream, long offset, long fragmentSize) throws IOException, InterruptedException { TextFormat format = getTextFormat(); RecordDefinition<T> record = getRecordDefinition(); Set<InputOption> options = offset == 0 ? INPUT_OPTS_HEAD : INPUT_OPTS_REST; InputStream source = decorate(stream, offset, fragmentSize); return new DecoratedInput(record.newInput(format.open(source), path, options), new StringOption(path)); }  final TextFormat getTextFormat(); final RecordDefinition<T> getRecordDefinition(); @Override final long getPreferredFragmentSize(); @Override final long getMinimumFragmentSize(); @Override ModelInput<T> createInput(
            Class<? extends T> dataType,
            String path,
            InputStream stream,
            long offset, long fragmentSize); @Override ModelOutput<T> createOutput(
            Class<? extends T> dataType,
            String path,
            OutputStream stream); }
AbstractTextStreamFormat extends ConfigurableBinaryStreamFormat<T> { @Override public ModelInput<T> createInput( Class<? extends T> dataType, String path, InputStream stream, long offset, long fragmentSize) throws IOException, InterruptedException { TextFormat format = getTextFormat(); RecordDefinition<T> record = getRecordDefinition(); Set<InputOption> options = offset == 0 ? INPUT_OPTS_HEAD : INPUT_OPTS_REST; InputStream source = decorate(stream, offset, fragmentSize); return new DecoratedInput(record.newInput(format.open(source), path, options), new StringOption(path)); }  final TextFormat getTextFormat(); final RecordDefinition<T> getRecordDefinition(); @Override final long getPreferredFragmentSize(); @Override final long getMinimumFragmentSize(); @Override ModelInput<T> createInput(
            Class<? extends T> dataType,
            String path,
            InputStream stream,
            long offset, long fragmentSize); @Override ModelOutput<T> createOutput(
            Class<? extends T> dataType,
            String path,
            OutputStream stream);  }