ReactiveDataConsumer implements AsyncDataConsumer, Publisher<ByteBuffer> { public void failed(final Exception cause) { exception = cause; flushToSubscriber(); }  void failed(final Exception cause); @Override void updateCapacity(final CapacityChannel capacityChannel); @Override void consume(final ByteBuffer byteBuffer); @Override void streamEnd(final List<? extends Header> trailers); @Override void releaseResources(); @Override void subscribe(final Subscriber<? super ByteBuffer> subscriber);  }
@Test public void testStreamThatEndsWithError() { final ReactiveDataConsumer consumer = new ReactiveDataConsumer(); final Single<List<Notification<ByteBuffer>>> single = Observable.fromPublisher(consumer) .materialize() .toList(); final Exception ex = new RuntimeException(); consumer.failed(ex); Assert.assertSame(ex, single.blockingGet().get(0).getError()); }