@Test public void cancelShouldTerminatePostDrain() throws Exception { BlockingIterator<Integer> iterator = source.toIterable().iterator(); assertTrue(source.isSubscribed()); source.onNext(1, 2); iterator.close(); assertThat("Unexpected item found.", iterator.next(), is(1)); assertThat("Unexpected item found.", iterator.next(), is(2)); 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();  }