@Test public void testConsumerError() { StreamSupplier<Integer> source = StreamSupplier.of(1, 2, 3, 4, 5, 6); StreamTransducer<Integer, Integer> transducer = StreamTransducer.create(new TransducerPipe<>()); StreamConsumerToList<Integer> consumer = StreamConsumerToList.create(); ExpectedException exception = new ExpectedException("Test Exception"); Throwable e = awaitException(source.transformWith(transducer) .streamTo(consumer .transformWith(decorate(promise -> promise.then(item -> item == 4 ? Promise.ofException(exception) : Promise.of(item)))))); assertSame(exception, e); assertEquals(asList(1, 2, 3, 4), consumer.getList()); assertClosedWithError(source); assertClosedWithError(consumer); assertClosedWithError(transducer); }
public static <I, O> StreamTransducer<I, O> create(Transducer<I, O, ?> transducer) { return new StreamTransducer<>(transducer); }
StreamTransducer implements StreamTransformer<I, O> { public static <I, O> StreamTransducer<I, O> create(Transducer<I, O, ?> transducer) { return new StreamTransducer<>(transducer); } }
StreamTransducer implements StreamTransformer<I, O> { public static <I, O> StreamTransducer<I, O> create(Transducer<I, O, ?> transducer) { return new StreamTransducer<>(transducer); } private  StreamTransducer(Transducer<I, O, ?> transducer); }
StreamTransducer implements StreamTransformer<I, O> { public static <I, O> StreamTransducer<I, O> create(Transducer<I, O, ?> transducer) { return new StreamTransducer<>(transducer); } private  StreamTransducer(Transducer<I, O, ?> transducer); static StreamTransducer<I, O> create(Transducer<I, O, ?> transducer); @Override StreamConsumer<I> getInput(); @Override StreamSupplier<O> getOutput(); }
StreamTransducer implements StreamTransformer<I, O> { public static <I, O> StreamTransducer<I, O> create(Transducer<I, O, ?> transducer) { return new StreamTransducer<>(transducer); } private  StreamTransducer(Transducer<I, O, ?> transducer); static StreamTransducer<I, O> create(Transducer<I, O, ?> transducer); @Override StreamConsumer<I> getInput(); @Override StreamSupplier<O> getOutput();  }