@Test public void input_splitter_trim_around() throws Exception { MockFormat format = format(3) .withInputSplitter(InputSplitters.byLineFeed()); int[][] data = { { 11, 12, 13, }, { 22, 22, 23, }, { 33, 32, 33, }, }; int objsz = 2 + (4 + 2) * 3 + 1; try (ModelInput<int[]> in = format.createInput(int[].class, "dummy", input(data), 1, objsz * 2 - 1)) { int[][] result = collect(3, in); assertThat(result, is(Arrays.copyOfRange(data, 1, 2))); } }
@Override public ModelInput<T> createInput( Class<? extends T> dataType, String path, InputStream stream, long offset, long fragmentSize) throws IOException, InterruptedException { JsonFormat<T> format = getJsonFormat(); Set<InputOption> options = offset == 0 ? INPUT_OPTS_HEAD : INPUT_OPTS_REST; InputStream source = decorate(stream, offset, fragmentSize); return new DecoratedInput(format.open(path, source, options), new StringOption(path)); }
AbstractJsonStreamFormat extends ConfigurableBinaryStreamFormat<T> { @Override public ModelInput<T> createInput( Class<? extends T> dataType, String path, InputStream stream, long offset, long fragmentSize) throws IOException, InterruptedException { JsonFormat<T> format = getJsonFormat(); Set<InputOption> options = offset == 0 ? INPUT_OPTS_HEAD : INPUT_OPTS_REST; InputStream source = decorate(stream, offset, fragmentSize); return new DecoratedInput(format.open(path, source, options), new StringOption(path)); } }
AbstractJsonStreamFormat extends ConfigurableBinaryStreamFormat<T> { @Override public ModelInput<T> createInput( Class<? extends T> dataType, String path, InputStream stream, long offset, long fragmentSize) throws IOException, InterruptedException { JsonFormat<T> format = getJsonFormat(); Set<InputOption> options = offset == 0 ? INPUT_OPTS_HEAD : INPUT_OPTS_REST; InputStream source = decorate(stream, offset, fragmentSize); return new DecoratedInput(format.open(path, source, options), new StringOption(path)); }  }
AbstractJsonStreamFormat extends ConfigurableBinaryStreamFormat<T> { @Override public ModelInput<T> createInput( Class<? extends T> dataType, String path, InputStream stream, long offset, long fragmentSize) throws IOException, InterruptedException { JsonFormat<T> format = getJsonFormat(); Set<InputOption> options = offset == 0 ? INPUT_OPTS_HEAD : INPUT_OPTS_REST; InputStream source = decorate(stream, offset, fragmentSize); return new DecoratedInput(format.open(path, source, options), new StringOption(path)); }  final JsonFormat<T> getJsonFormat(); @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); }
AbstractJsonStreamFormat extends ConfigurableBinaryStreamFormat<T> { @Override public ModelInput<T> createInput( Class<? extends T> dataType, String path, InputStream stream, long offset, long fragmentSize) throws IOException, InterruptedException { JsonFormat<T> format = getJsonFormat(); Set<InputOption> options = offset == 0 ? INPUT_OPTS_HEAD : INPUT_OPTS_REST; InputStream source = decorate(stream, offset, fragmentSize); return new DecoratedInput(format.open(path, source, options), new StringOption(path)); }  final JsonFormat<T> getJsonFormat(); @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);  }