@Test public void breakReentrantRequestN() throws Throwable { AtomicInteger count = new AtomicInteger(); AtomicBoolean complete = new AtomicBoolean(); AtomicReference<Throwable> error = new AtomicReference<>(); when(inputStream.available()).thenReturn(1); when(inputStream.read(any(), anyInt(), anyInt())).then(inv -> { if (count.incrementAndGet() > 1_000) { return -1; } return 1; }); toSource(pub).subscribe(new Subscriber<byte[]>() { private Subscription s; @Override public void onSubscribe(final Subscription s) { this.s = s; s.request(1); } @Override public void onNext(final byte[] b) { s.request(1); } @Override public void onError(final Throwable t) { error.set(t); } @Override public void onComplete() { complete.set(true); } }); if (error.get() != null) { throw error.get(); } assertThat(complete.get(), equalTo(true)); }
@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);  }