@Test public void testNormal() { Subscriber<Object> o = TestHelper.mockSubscriber(); final List<Integer> list = Arrays.asList(1, 2, 3); Function<Integer, List<Integer>> func = new Function<Integer, List<Integer>>() { @Override public List<Integer> apply(Integer t1) { return list; } }; BiFunction<Integer, Integer, Integer> resFunc = new BiFunction<Integer, Integer, Integer>() { @Override public Integer apply(Integer t1, Integer t2) { return t1 | t2; } }; List<Integer> source = Arrays.asList(16, 32, 64); Flowable.fromIterable(source).flatMapIterable(func, resFunc).subscribe(o); for (Integer s : source) { for (Integer v : list) { verify(o).onNext(s | v); } } verify(o).onComplete(); verify(o, never()).onError(any(Throwable.class)); }
public static <T, U> FlowableSubscriber<T> subscribe(Subscriber<? super U> s, Function<? super T, ? extends Publisher<? extends U>> mapper, boolean delayErrors, int maxConcurrency, int bufferSize) { return new MergeSubscriber<T, U>(s, mapper, delayErrors, maxConcurrency, bufferSize); }
FlowableFlatMap extends AbstractFlowableWithUpstream<T, U> { public static <T, U> FlowableSubscriber<T> subscribe(Subscriber<? super U> s, Function<? super T, ? extends Publisher<? extends U>> mapper, boolean delayErrors, int maxConcurrency, int bufferSize) { return new MergeSubscriber<T, U>(s, mapper, delayErrors, maxConcurrency, bufferSize); } }
FlowableFlatMap extends AbstractFlowableWithUpstream<T, U> { public static <T, U> FlowableSubscriber<T> subscribe(Subscriber<? super U> s, Function<? super T, ? extends Publisher<? extends U>> mapper, boolean delayErrors, int maxConcurrency, int bufferSize) { return new MergeSubscriber<T, U>(s, mapper, delayErrors, maxConcurrency, bufferSize); } FlowableFlatMap(Flowable<T> source,
            Function<? super T, ? extends Publisher<? extends U>> mapper,
            boolean delayErrors, int maxConcurrency, int bufferSize); }
FlowableFlatMap extends AbstractFlowableWithUpstream<T, U> { public static <T, U> FlowableSubscriber<T> subscribe(Subscriber<? super U> s, Function<? super T, ? extends Publisher<? extends U>> mapper, boolean delayErrors, int maxConcurrency, int bufferSize) { return new MergeSubscriber<T, U>(s, mapper, delayErrors, maxConcurrency, bufferSize); } FlowableFlatMap(Flowable<T> source,
            Function<? super T, ? extends Publisher<? extends U>> mapper,
            boolean delayErrors, int maxConcurrency, int bufferSize); static FlowableSubscriber<T> subscribe(Subscriber<? super U> s,
            Function<? super T, ? extends Publisher<? extends U>> mapper,
            boolean delayErrors, int maxConcurrency, int bufferSize); }
FlowableFlatMap extends AbstractFlowableWithUpstream<T, U> { public static <T, U> FlowableSubscriber<T> subscribe(Subscriber<? super U> s, Function<? super T, ? extends Publisher<? extends U>> mapper, boolean delayErrors, int maxConcurrency, int bufferSize) { return new MergeSubscriber<T, U>(s, mapper, delayErrors, maxConcurrency, bufferSize); } FlowableFlatMap(Flowable<T> source,
            Function<? super T, ? extends Publisher<? extends U>> mapper,
            boolean delayErrors, int maxConcurrency, int bufferSize); static FlowableSubscriber<T> subscribe(Subscriber<? super U> s,
            Function<? super T, ? extends Publisher<? extends U>> mapper,
            boolean delayErrors, int maxConcurrency, int bufferSize);  }