@Test public void fusionWithConcatMap() { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); Flowable.fromIterable(Arrays.asList(1, 2, 3, 4)).concatMap( new Function<Integer, Flowable <Integer>>() { @Override public Flowable<Integer> apply(Integer v) { return Flowable.range(v, 2); } }).subscribe(ts); ts.assertValues(1, 2, 2, 3, 3, 4, 4, 5); ts.assertNoErrors(); ts.assertComplete(); }
public static <T> void subscribe(Subscriber<? super T> s, Iterator<? extends T> it) { boolean hasNext; try { hasNext = it.hasNext(); } catch (Throwable e) { Exceptions.throwIfFatal(e); EmptySubscription.error(e, s); return; } if (!hasNext) { EmptySubscription.complete(s); return; } if (s instanceof ConditionalSubscriber) { s.onSubscribe(new IteratorConditionalSubscription<T>( (ConditionalSubscriber<? super T>)s, it)); } else { s.onSubscribe(new IteratorSubscription<T>(s, it)); } }
FlowableFromIterable extends Flowable<T> { public static <T> void subscribe(Subscriber<? super T> s, Iterator<? extends T> it) { boolean hasNext; try { hasNext = it.hasNext(); } catch (Throwable e) { Exceptions.throwIfFatal(e); EmptySubscription.error(e, s); return; } if (!hasNext) { EmptySubscription.complete(s); return; } if (s instanceof ConditionalSubscriber) { s.onSubscribe(new IteratorConditionalSubscription<T>( (ConditionalSubscriber<? super T>)s, it)); } else { s.onSubscribe(new IteratorSubscription<T>(s, it)); } } }
FlowableFromIterable extends Flowable<T> { public static <T> void subscribe(Subscriber<? super T> s, Iterator<? extends T> it) { boolean hasNext; try { hasNext = it.hasNext(); } catch (Throwable e) { Exceptions.throwIfFatal(e); EmptySubscription.error(e, s); return; } if (!hasNext) { EmptySubscription.complete(s); return; } if (s instanceof ConditionalSubscriber) { s.onSubscribe(new IteratorConditionalSubscription<T>( (ConditionalSubscriber<? super T>)s, it)); } else { s.onSubscribe(new IteratorSubscription<T>(s, it)); } } FlowableFromIterable(Iterable<? extends T> source); }
FlowableFromIterable extends Flowable<T> { public static <T> void subscribe(Subscriber<? super T> s, Iterator<? extends T> it) { boolean hasNext; try { hasNext = it.hasNext(); } catch (Throwable e) { Exceptions.throwIfFatal(e); EmptySubscription.error(e, s); return; } if (!hasNext) { EmptySubscription.complete(s); return; } if (s instanceof ConditionalSubscriber) { s.onSubscribe(new IteratorConditionalSubscription<T>( (ConditionalSubscriber<? super T>)s, it)); } else { s.onSubscribe(new IteratorSubscription<T>(s, it)); } } FlowableFromIterable(Iterable<? extends T> source); @Override void subscribeActual(Subscriber<? super T> s); static void subscribe(Subscriber<? super T> s, Iterator<? extends T> it); }
FlowableFromIterable extends Flowable<T> { public static <T> void subscribe(Subscriber<? super T> s, Iterator<? extends T> it) { boolean hasNext; try { hasNext = it.hasNext(); } catch (Throwable e) { Exceptions.throwIfFatal(e); EmptySubscription.error(e, s); return; } if (!hasNext) { EmptySubscription.complete(s); return; } if (s instanceof ConditionalSubscriber) { s.onSubscribe(new IteratorConditionalSubscription<T>( (ConditionalSubscriber<? super T>)s, it)); } else { s.onSubscribe(new IteratorSubscription<T>(s, it)); } } FlowableFromIterable(Iterable<? extends T> source); @Override void subscribeActual(Subscriber<? super T> s); static void subscribe(Subscriber<? super T> s, Iterator<? extends T> it);  }