@Test public void testBufferedConsumer() { StreamSupplier<Integer> supplier = StreamSupplier.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); StreamConsumerToList<Integer> consumer = StreamConsumerToList.create(); StreamBuffer<Integer> buffer = StreamBuffer.create(1, 2); await(supplier.streamTo(consumer.transformWith(buffer).transformWith(randomlySuspending()))); assertEquals(asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10), consumer.getList()); assertEndOfStream(supplier, consumer); assertEndOfStream(buffer); }
public static <T> StreamBuffer<T> create(int bufferMinSize, int bufferMaxSize) { return new StreamBuffer<T>(bufferMinSize, bufferMaxSize); }
StreamBuffer implements StreamTransformer<T, T> { public static <T> StreamBuffer<T> create(int bufferMinSize, int bufferMaxSize) { return new StreamBuffer<T>(bufferMinSize, bufferMaxSize); } }
StreamBuffer implements StreamTransformer<T, T> { public static <T> StreamBuffer<T> create(int bufferMinSize, int bufferMaxSize) { return new StreamBuffer<T>(bufferMinSize, bufferMaxSize); } private  StreamBuffer(int bufferMinSize, int bufferMaxSize); }
StreamBuffer implements StreamTransformer<T, T> { public static <T> StreamBuffer<T> create(int bufferMinSize, int bufferMaxSize) { return new StreamBuffer<T>(bufferMinSize, bufferMaxSize); } private  StreamBuffer(int bufferMinSize, int bufferMaxSize); static StreamBuffer<T> create(int bufferMinSize, int bufferMaxSize); boolean isSaturated(); boolean isExhausted(); boolean isEmpty(); int size(); @Override StreamConsumer<T> getInput(); @Override StreamSupplier<T> getOutput(); }
StreamBuffer implements StreamTransformer<T, T> { public static <T> StreamBuffer<T> create(int bufferMinSize, int bufferMaxSize) { return new StreamBuffer<T>(bufferMinSize, bufferMaxSize); } private  StreamBuffer(int bufferMinSize, int bufferMaxSize); static StreamBuffer<T> create(int bufferMinSize, int bufferMaxSize); boolean isSaturated(); boolean isExhausted(); boolean isEmpty(); int size(); @Override StreamConsumer<T> getInput(); @Override StreamSupplier<T> getOutput();  }