@Test public void cancelAfterSecondHasNext() { TestSubscriber<Integer> ts = new TestSubscriber<>(Long.MAX_VALUE); Multi.just(1) .flatMapIterable(v -> () -> { return new Iterator<Integer>() { int count; @Override public boolean hasNext() { if (++count == 2) { ts.cancel(); } return true; } @Override public Integer next() { return 1; } }; }) .subscribe(ts); ts.assertValuesOnly(1); }
@Override public void subscribe(Flow.Subscriber<? super R> subscriber) { source.subscribe(new FlatMapIterableSubscriber<>(subscriber, mapper, prefetch)); }
MultiFlatMapIterable implements Multi<R> { @Override public void subscribe(Flow.Subscriber<? super R> subscriber) { source.subscribe(new FlatMapIterableSubscriber<>(subscriber, mapper, prefetch)); } }
MultiFlatMapIterable implements Multi<R> { @Override public void subscribe(Flow.Subscriber<? super R> subscriber) { source.subscribe(new FlatMapIterableSubscriber<>(subscriber, mapper, prefetch)); } MultiFlatMapIterable(Multi<T> source,
                         Function<? super T, ? extends Iterable<? extends R>> mapper,
                         int prefetch); }
MultiFlatMapIterable implements Multi<R> { @Override public void subscribe(Flow.Subscriber<? super R> subscriber) { source.subscribe(new FlatMapIterableSubscriber<>(subscriber, mapper, prefetch)); } MultiFlatMapIterable(Multi<T> source,
                         Function<? super T, ? extends Iterable<? extends R>> mapper,
                         int prefetch); @Override void subscribe(Flow.Subscriber<? super R> subscriber); }
MultiFlatMapIterable implements Multi<R> { @Override public void subscribe(Flow.Subscriber<? super R> subscriber) { source.subscribe(new FlatMapIterableSubscriber<>(subscriber, mapper, prefetch)); } MultiFlatMapIterable(Multi<T> source,
                         Function<? super T, ? extends Iterable<? extends R>> mapper,
                         int prefetch); @Override void subscribe(Flow.Subscriber<? super R> subscriber);  }