@Test @SuppressWarnings("unchecked") public void streamClosedAndErrorOnDeliveryError() throws Exception { initChunkedStream(smallBuff, of(10), of(10)); Subscriber sub = mock(Subscriber.class); doAnswer(inv -> { Subscription s = inv.getArgument(0); s.request(1); return null; }).when(sub).onSubscribe(any()); doThrow(DELIBERATE_EXCEPTION).when(sub).onNext(any()); toSource(pub).subscribe(sub); verify(inputStream).close(); verify(sub, never()).onComplete(); verify(sub).onError(DELIBERATE_EXCEPTION); }
@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);  }