@Test public void iteratorHasNextCrash2ndCall() { TestSubscriber<Integer> ts = new TestSubscriber<>(); Multi.create(() -> new Iterator<Integer>() { int calls; @Override public boolean hasNext() { if (++calls == 2) { throw new IllegalArgumentException(); } return true; } @Override public Integer next() { return 1; } }) .subscribe(ts); ts.request1(); assertEquals(ts.getItems(), Collections.singleton(1)); assertThat(ts.isComplete(), is(false)); assertThat(ts.getLastError(), instanceOf(IllegalArgumentException.class)); }
@Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); Iterator<T> iterator; boolean hasFirst; try { iterator = iterable.iterator(); hasFirst = iterator.hasNext(); } catch (Throwable ex) { subscriber.onSubscribe(EmptySubscription.INSTANCE); subscriber.onError(ex); return; } if (!hasFirst) { subscriber.onSubscribe(EmptySubscription.INSTANCE); subscriber.onComplete(); return; } subscriber.onSubscribe(new IteratorSubscription<>(subscriber, iterator)); }
MultiFromIterable implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); Iterator<T> iterator; boolean hasFirst; try { iterator = iterable.iterator(); hasFirst = iterator.hasNext(); } catch (Throwable ex) { subscriber.onSubscribe(EmptySubscription.INSTANCE); subscriber.onError(ex); return; } if (!hasFirst) { subscriber.onSubscribe(EmptySubscription.INSTANCE); subscriber.onComplete(); return; } subscriber.onSubscribe(new IteratorSubscription<>(subscriber, iterator)); } }
MultiFromIterable implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); Iterator<T> iterator; boolean hasFirst; try { iterator = iterable.iterator(); hasFirst = iterator.hasNext(); } catch (Throwable ex) { subscriber.onSubscribe(EmptySubscription.INSTANCE); subscriber.onError(ex); return; } if (!hasFirst) { subscriber.onSubscribe(EmptySubscription.INSTANCE); subscriber.onComplete(); return; } subscriber.onSubscribe(new IteratorSubscription<>(subscriber, iterator)); } MultiFromIterable(Iterable<T> iterable); }
MultiFromIterable implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); Iterator<T> iterator; boolean hasFirst; try { iterator = iterable.iterator(); hasFirst = iterator.hasNext(); } catch (Throwable ex) { subscriber.onSubscribe(EmptySubscription.INSTANCE); subscriber.onError(ex); return; } if (!hasFirst) { subscriber.onSubscribe(EmptySubscription.INSTANCE); subscriber.onComplete(); return; } subscriber.onSubscribe(new IteratorSubscription<>(subscriber, iterator)); } MultiFromIterable(Iterable<T> iterable); @Override void subscribe(Flow.Subscriber<? super T> subscriber); }
MultiFromIterable implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); Iterator<T> iterator; boolean hasFirst; try { iterator = iterable.iterator(); hasFirst = iterator.hasNext(); } catch (Throwable ex) { subscriber.onSubscribe(EmptySubscription.INSTANCE); subscriber.onError(ex); return; } if (!hasFirst) { subscriber.onSubscribe(EmptySubscription.INSTANCE); subscriber.onComplete(); return; } subscriber.onSubscribe(new IteratorSubscription<>(subscriber, iterator)); } MultiFromIterable(Iterable<T> iterable); @Override void subscribe(Flow.Subscriber<? super T> subscriber);  }