@Test public void testConsumerError() { ExpectedException expectedException = new ExpectedException(); 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); Throwable exception = awaitException(supplier.streamTo(consumer .transformWith(buffer) .transformWith(decorate(promise -> promise.then( item -> item == 5 ? Promise.ofException(expectedException) : Promise.of(item)))) .transformWith(randomlySuspending()))); assertSame(expectedException, exception); assertEquals(asList(1, 2, 3, 4, 5), consumer.getList()); assertClosedWithError(expectedException, supplier, consumer); assertClosedWithError(expectedException, 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();  }