@Test public void delayOnNextThenComplete() { Iterator<Integer> iterator = source.toIterable(2).iterator(); TestSubscription subscription = new TestSubscription(); source.onSubscribe(subscription); assertTrue(source.isSubscribed()); assertThat(subscription.requested(), is((long) 2)); source.onNext(1); verifyNextIs(iterator, 1); source.onNext(2); verifyNextIs(iterator, 2); source.onComplete(); assertThat("Item not expected but found.", iterator.hasNext(), is(false)); }
@Override public BlockingIterator<T> iterator() { SubscriberAndIterator<T> subscriberAndIterator = new SubscriberAndIterator<>(queueCapacityHint); original.subscribeInternal(subscriberAndIterator); return subscriberAndIterator; }
PublisherAsBlockingIterable implements BlockingIterable<T> { @Override public BlockingIterator<T> iterator() { SubscriberAndIterator<T> subscriberAndIterator = new SubscriberAndIterator<>(queueCapacityHint); original.subscribeInternal(subscriberAndIterator); return subscriberAndIterator; } }
PublisherAsBlockingIterable implements BlockingIterable<T> { @Override public BlockingIterator<T> iterator() { SubscriberAndIterator<T> subscriberAndIterator = new SubscriberAndIterator<>(queueCapacityHint); original.subscribeInternal(subscriberAndIterator); return subscriberAndIterator; } PublisherAsBlockingIterable(final Publisher<T> original);   PublisherAsBlockingIterable(final Publisher<T> original, int queueCapacityHint); }
PublisherAsBlockingIterable implements BlockingIterable<T> { @Override public BlockingIterator<T> iterator() { SubscriberAndIterator<T> subscriberAndIterator = new SubscriberAndIterator<>(queueCapacityHint); original.subscribeInternal(subscriberAndIterator); return subscriberAndIterator; } PublisherAsBlockingIterable(final Publisher<T> original);   PublisherAsBlockingIterable(final Publisher<T> original, int queueCapacityHint); @Override BlockingIterator<T> iterator(); }
PublisherAsBlockingIterable implements BlockingIterable<T> { @Override public BlockingIterator<T> iterator() { SubscriberAndIterator<T> subscriberAndIterator = new SubscriberAndIterator<>(queueCapacityHint); original.subscribeInternal(subscriberAndIterator); return subscriberAndIterator; } PublisherAsBlockingIterable(final Publisher<T> original);   PublisherAsBlockingIterable(final Publisher<T> original, int queueCapacityHint); @Override BlockingIterator<T> iterator();  }