@Test public void errorEmittedIsThrown() { DeliberateException de = new DeliberateException(); Iterator<Integer> iterator = Publisher.<Integer>failed(de).toIterable().iterator(); assertThat("Item expected but not found.", iterator.hasNext(), is(true)); expected.expect(sameInstance(de)); iterator.next(); }
@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();  }