@Test @SuppressWarnings("unchecked") public void streamClosedAndErrorOnDeliveryErrorOnce() throws Exception { initChunkedStream(smallBuff, ofAll(10), ofAll(10)); Subscriber sub = mock(Subscriber.class); AtomicReference<Subscription> subRef = new AtomicReference<>(); doAnswer(inv -> { Subscription s = inv.getArgument(0); subRef.set(s); s.request(1); return null; }).when(sub).onSubscribe(any()); doThrow(DELIBERATE_EXCEPTION).when(sub).onNext(any()); toSource(pub).subscribe(sub); subRef.get().request(1); verify(sub, never()).onComplete(); verify(sub).onError(DELIBERATE_EXCEPTION); verify(inputStream).close(); }
@Override public void subscribe(final Subscriber<? super byte[]> subscriber) { subscribeInternal(subscriber); }
FromInputStreamPublisher extends Publisher<byte[]> implements PublisherSource<byte[]> { @Override public void subscribe(final Subscriber<? super byte[]> subscriber) { subscribeInternal(subscriber); } }
FromInputStreamPublisher extends Publisher<byte[]> implements PublisherSource<byte[]> { @Override public void subscribe(final Subscriber<? super byte[]> subscriber) { subscribeInternal(subscriber); } FromInputStreamPublisher(final InputStream stream); }
FromInputStreamPublisher extends Publisher<byte[]> implements PublisherSource<byte[]> { @Override public void subscribe(final Subscriber<? super byte[]> subscriber) { subscribeInternal(subscriber); } FromInputStreamPublisher(final InputStream stream); @Override void subscribe(final Subscriber<? super byte[]> subscriber); }
FromInputStreamPublisher extends Publisher<byte[]> implements PublisherSource<byte[]> { @Override public void subscribe(final Subscriber<? super byte[]> subscriber) { subscribeInternal(subscriber); } FromInputStreamPublisher(final InputStream stream); @Override void subscribe(final Subscriber<? super byte[]> subscriber);  }