@Test public void consumeSimpleStream() throws Exception { initChunkedStream(smallBuff, of(10, 0), of(10, 0)); toSource(pub).subscribe(sub1); sub1.request(1); assertThat(sub1.takeItems(), contains(smallBuff)); sub1.request(1); assertThat(sub1.takeTerminal(), is(complete())); }
@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);  }