@Test public void errorEmittedIsThrownAfterEmittingAllItems() { DeliberateException de = new DeliberateException(); Iterator<Integer> iterator = from(1, 2, 3, 4) .concat(Publisher.failed(de)).toIterable().iterator(); List<Integer> result = new ArrayList<>(4); try { while (iterator.hasNext()) { result.add(iterator.next()); } fail("Exception expected but not thrown."); } catch (DeliberateException e) { assertThat("Unexpected exception.", e, sameInstance(de)); assertThat("Unexpected result.", result, contains(1, 2, 3, 4)); } }
@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();  }