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);  }
@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)); }
@Test public void testCollectionFunctionThrows() { Subscriber<Object> o = TestHelper.mockSubscriber(); Function<Integer, List<Integer>> func = new Function<Integer, List<Integer>>() { @Override public List<Integer> apply(Integer t1) { throw new TestException(); } }; 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); verify(o, never()).onComplete(); verify(o, never()).onNext(any()); verify(o).onError(any(TestException.class)); }
@Test public void testResultFunctionThrows() { 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) { throw new TestException(); } }; List<Integer> source = Arrays.asList(16, 32, 64); Flowable.fromIterable(source).flatMapIterable(func, resFunc).subscribe(o); verify(o, never()).onComplete(); verify(o, never()).onNext(any()); verify(o).onError(any(TestException.class)); }
@Test public void testMergeError() { Subscriber<Object> o = TestHelper.mockSubscriber(); Function<Integer, Flowable<Integer>> func = new Function<Integer, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Integer t1) { return Flowable.error(new TestException()); } }; 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).flatMap(func, resFunc).subscribe(o); verify(o, never()).onComplete(); verify(o, never()).onNext(any()); verify(o).onError(any(TestException.class)); }
@Test public void testFlatMapTransformsNormal() { Flowable<Integer> onNext = Flowable.fromIterable(Arrays.asList(1, 2, 3)); Flowable<Integer> onComplete = Flowable.fromIterable(Arrays.asList(4)); Flowable<Integer> onError = Flowable.fromIterable(Arrays.asList(5)); Flowable<Integer> source = Flowable.fromIterable(Arrays.asList(10, 20, 30)); Subscriber<Object> o = TestHelper.mockSubscriber(); source.flatMap(just(onNext), just(onError), just0(onComplete)).subscribe(o); verify(o, times(3)).onNext(1); verify(o, times(3)).onNext(2); verify(o, times(3)).onNext(3); verify(o).onNext(4); verify(o).onComplete(); verify(o, never()).onNext(5); verify(o, never()).onError(any(Throwable.class)); }
@Test public void testFlatMapTransformsException() { Flowable<Integer> onNext = Flowable.fromIterable(Arrays.asList(1, 2, 3)); Flowable<Integer> onComplete = Flowable.fromIterable(Arrays.asList(4)); Flowable<Integer> onError = Flowable.fromIterable(Arrays.asList(5)); Flowable<Integer> source = Flowable.concat( Flowable.fromIterable(Arrays.asList(10, 20, 30)), Flowable.<Integer> error(new RuntimeException("Forced failure!")) ); Subscriber<Object> o = TestHelper.mockSubscriber(); source.flatMap(just(onNext), just(onError), just0(onComplete)).subscribe(o); verify(o, times(3)).onNext(1); verify(o, times(3)).onNext(2); verify(o, times(3)).onNext(3); verify(o).onNext(5); verify(o).onComplete(); verify(o, never()).onNext(4); verify(o, never()).onError(any(Throwable.class)); }
@Test public void testFlatMapTransformsOnNextFuncThrows() { Flowable<Integer> onComplete = Flowable.fromIterable(Arrays.asList(4)); Flowable<Integer> onError = Flowable.fromIterable(Arrays.asList(5)); Flowable<Integer> source = Flowable.fromIterable(Arrays.asList(10, 20, 30)); Subscriber<Object> o = TestHelper.mockSubscriber(); source.flatMap(funcThrow(1, onError), just(onError), just0(onComplete)).subscribe(o); verify(o).onError(any(TestException.class)); verify(o, never()).onNext(any()); verify(o, never()).onComplete(); }
@Test public void testFlatMapTransformsOnErrorFuncThrows() { Flowable<Integer> onNext = Flowable.fromIterable(Arrays.asList(1, 2, 3)); Flowable<Integer> onComplete = Flowable.fromIterable(Arrays.asList(4)); Flowable<Integer> onError = Flowable.fromIterable(Arrays.asList(5)); Flowable<Integer> source = Flowable.error(new TestException()); Subscriber<Object> o = TestHelper.mockSubscriber(); source.flatMap(just(onNext), funcThrow((Throwable) null, onError), just0(onComplete)).subscribe(o); verify(o).onError(any(CompositeException.class)); verify(o, never()).onNext(any()); verify(o, never()).onComplete(); }
@Test public void testFlatMapTransformsOnCompletedFuncThrows() { Flowable<Integer> onNext = Flowable.fromIterable(Arrays.asList(1, 2, 3)); Flowable<Integer> onComplete = Flowable.fromIterable(Arrays.asList(4)); Flowable<Integer> onError = Flowable.fromIterable(Arrays.asList(5)); Flowable<Integer> source = Flowable.fromIterable(Arrays.<Integer> asList()); Subscriber<Object> o = TestHelper.mockSubscriber(); source.flatMap(just(onNext), just(onError), funcThrow0(onComplete)).subscribe(o); verify(o).onError(any(TestException.class)); verify(o, never()).onNext(any()); verify(o, never()).onComplete(); }
@Test public void testFlatMapTransformsMergeException() { Flowable<Integer> onNext = Flowable.error(new TestException()); Flowable<Integer> onComplete = Flowable.fromIterable(Arrays.asList(4)); Flowable<Integer> onError = Flowable.fromIterable(Arrays.asList(5)); Flowable<Integer> source = Flowable.fromIterable(Arrays.asList(10, 20, 30)); Subscriber<Object> o = TestHelper.mockSubscriber(); source.flatMap(just(onNext), just(onError), funcThrow0(onComplete)).subscribe(o); verify(o).onError(any(TestException.class)); verify(o, never()).onNext(any()); verify(o, never()).onComplete(); }
@Test public void testFlatMapMaxConcurrent() { final int m = 4; final AtomicInteger subscriptionCount = new AtomicInteger(); Flowable<Integer> source = Flowable.range(1, 10) .flatMap(new Function<Integer, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Integer t1) { return composer(Flowable.range(t1 * 10, 2), subscriptionCount, m) .subscribeOn(Schedulers.computation()); } }, m); TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); source.subscribe(ts); ts.awaitTerminalEvent(); ts.assertNoErrors(); Set<Integer> expected = new HashSet<Integer>(Arrays.asList( 10, 11, 20, 21, 30, 31, 40, 41, 50, 51, 60, 61, 70, 71, 80, 81, 90, 91, 100, 101 )); Assert.assertEquals(expected.size(), ts.valueCount()); Assert.assertTrue(expected.containsAll(ts.values())); }
@Test public void testFlatMapSelectorMaxConcurrent() { final int m = 4; final AtomicInteger subscriptionCount = new AtomicInteger(); Flowable<Integer> source = Flowable.range(1, 10) .flatMap(new Function<Integer, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Integer t1) { return composer(Flowable.range(t1 * 10, 2), subscriptionCount, m) .subscribeOn(Schedulers.computation()); } }, new BiFunction<Integer, Integer, Integer>() { @Override public Integer apply(Integer t1, Integer t2) { return t1 * 1000 + t2; } }, m); TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); source.subscribe(ts); ts.awaitTerminalEvent(); ts.assertNoErrors(); Set<Integer> expected = new HashSet<Integer>(Arrays.asList( 1010, 1011, 2020, 2021, 3030, 3031, 4040, 4041, 5050, 5051, 6060, 6061, 7070, 7071, 8080, 8081, 9090, 9091, 10100, 10101 )); Assert.assertEquals(expected.size(), ts.valueCount()); System.out.println("--> testFlatMapSelectorMaxConcurrent: " + ts.values()); Assert.assertTrue(expected.containsAll(ts.values())); }
@Test public void testFlatMapTransformsMaxConcurrentNormal() { final int m = 2; final AtomicInteger subscriptionCount = new AtomicInteger(); Flowable<Integer> onNext = composer( Flowable.fromIterable(Arrays.asList(1, 2, 3)) .observeOn(Schedulers.computation()) , subscriptionCount, m) .subscribeOn(Schedulers.computation()) ; Flowable<Integer> onComplete = composer(Flowable.fromIterable(Arrays.asList(4)), subscriptionCount, m) .subscribeOn(Schedulers.computation()); Flowable<Integer> onError = Flowable.fromIterable(Arrays.asList(5)); Flowable<Integer> source = Flowable.fromIterable(Arrays.asList(10, 20, 30)); Subscriber<Object> o = TestHelper.mockSubscriber(); TestSubscriber<Object> ts = new TestSubscriber<Object>(o); Function<Integer, Flowable<Integer>> just = just(onNext); Function<Throwable, Flowable<Integer>> just2 = just(onError); Callable<Flowable<Integer>> just0 = just0(onComplete); source.flatMap(just, just2, just0, m).subscribe(ts); ts.awaitTerminalEvent(1, TimeUnit.SECONDS); ts.assertNoErrors(); ts.assertTerminated(); verify(o, times(3)).onNext(1); verify(o, times(3)).onNext(2); verify(o, times(3)).onNext(3); verify(o).onNext(4); verify(o).onComplete(); verify(o, never()).onNext(5); verify(o, never()).onError(any(Throwable.class)); }
@Ignore("Don't care for any reordering") @Test(timeout = 10000) public void flatMapRangeAsyncLoop() { for (int i = 0; i < 2000; i++) { if (i % 10 == 0) { System.out.println("flatMapRangeAsyncLoop > " + i); } TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); Flowable.range(0, 1000) .flatMap(new Function<Integer, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Integer t) { return Flowable.just(t); } }) .observeOn(Schedulers.computation()) .subscribe(ts); ts.awaitTerminalEvent(2500, TimeUnit.MILLISECONDS); if (ts.completions() == 0) { System.out.println(ts.valueCount()); } ts.assertTerminated(); ts.assertNoErrors(); List<Integer> list = ts.values(); assertEquals(1000, list.size()); boolean f = false; for (int j = 0; j < list.size(); j++) { if (list.get(j) != j) { System.out.println(j + " " + list.get(j)); f = true; } } if (f) { Assert.fail("Results are out of order!"); } } }
@Test(timeout = 30000) public void flatMapRangeMixedAsyncLoop() { for (int i = 0; i < 2000; i++) { if (i % 10 == 0) { System.out.println("flatMapRangeAsyncLoop > " + i); } TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); Flowable.range(0, 1000) .flatMap(new Function<Integer, Flowable<Integer>>() { final Random rnd = new Random(); @Override public Flowable<Integer> apply(Integer t) { Flowable<Integer> r = Flowable.just(t); if (rnd.nextBoolean()) { r = r.hide(); } return r; } }) .observeOn(Schedulers.computation()) .subscribe(ts); ts.awaitTerminalEvent(2500, TimeUnit.MILLISECONDS); if (ts.completions() == 0) { System.out.println(ts.valueCount()); } ts.assertTerminated(); ts.assertNoErrors(); List<Integer> list = ts.values(); if (list.size() < 1000) { Set<Integer> set = new HashSet<Integer>(list); for (int j = 0; j < 1000; j++) { if (!set.contains(j)) { System.out.println(j + " missing"); } } } assertEquals(1000, list.size()); } }
@Test public void flatMapIntPassthruAsync() { for (int i = 0;i < 1000; i++) { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); Flowable.range(1, 1000).flatMap(new Function<Integer, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Integer t) { return Flowable.just(1).subscribeOn(Schedulers.computation()); } }).subscribe(ts); ts.awaitTerminalEvent(5, TimeUnit.SECONDS); ts.assertNoErrors(); ts.assertComplete(); ts.assertValueCount(1000); } }
@Test public void flatMapTwoNestedSync() { for (final int n : new int[] { 1, 1000, 1000000 }) { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); Flowable.just(1, 2).flatMap(new Function<Integer, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Integer t) { return Flowable.range(1, n); } }).subscribe(ts); System.out.println("flatMapTwoNestedSync >> @ " + n); ts.assertNoErrors(); ts.assertComplete(); ts.assertValueCount(n * 2); } }
@Test public void justEmptyMixture() { TestSubscriber<Integer> ts = TestSubscriber.create(); Flowable.range(0, 4 * Flowable.bufferSize()) .flatMap(new Function<Integer, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Integer v) { return (v & 1) == 0 ? Flowable.<Integer>empty() : Flowable.just(v); } }) .subscribe(ts); ts.assertValueCount(2 * Flowable.bufferSize()); ts.assertNoErrors(); ts.assertComplete(); int j = 1; for (Integer v : ts.values()) { Assert.assertEquals(j, v.intValue()); j += 2; } }
@Test public void rangeEmptyMixture() { TestSubscriber<Integer> ts = TestSubscriber.create(); Flowable.range(0, 4 * Flowable.bufferSize()) .flatMap(new Function<Integer, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Integer v) { return (v & 1) == 0 ? Flowable.<Integer>empty() : Flowable.range(v, 2); } }) .subscribe(ts); ts.assertValueCount(4 * Flowable.bufferSize()); ts.assertNoErrors(); ts.assertComplete(); int j = 1; List<Integer> list = ts.values(); for (int i = 0; i < list.size(); i += 2) { Assert.assertEquals(j, list.get(i).intValue()); Assert.assertEquals(j + 1, list.get(i + 1).intValue()); j += 2; } }
@Test public void justEmptyMixtureMaxConcurrent() { TestSubscriber<Integer> ts = TestSubscriber.create(); Flowable.range(0, 4 * Flowable.bufferSize()) .flatMap(new Function<Integer, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Integer v) { return (v & 1) == 0 ? Flowable.<Integer>empty() : Flowable.just(v); } }, 16) .subscribe(ts); ts.assertValueCount(2 * Flowable.bufferSize()); ts.assertNoErrors(); ts.assertComplete(); int j = 1; for (Integer v : ts.values()) { Assert.assertEquals(j, v.intValue()); j += 2; } }
@Test public void rangeEmptyMixtureMaxConcurrent() { TestSubscriber<Integer> ts = TestSubscriber.create(); Flowable.range(0, 4 * Flowable.bufferSize()) .flatMap(new Function<Integer, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Integer v) { return (v & 1) == 0 ? Flowable.<Integer>empty() : Flowable.range(v, 2); } }, 16) .subscribe(ts); ts.assertValueCount(4 * Flowable.bufferSize()); ts.assertNoErrors(); ts.assertComplete(); int j = 1; List<Integer> list = ts.values(); for (int i = 0; i < list.size(); i += 2) { Assert.assertEquals(j, list.get(i).intValue()); Assert.assertEquals(j + 1, list.get(i + 1).intValue()); j += 2; } }
@Test public void castCrashUnsubscribes() { PublishProcessor<Integer> pp = PublishProcessor.create(); TestSubscriber<Integer> ts = TestSubscriber.create(); pp.flatMap(new Function<Integer, Publisher<Integer>>() { @Override public Publisher<Integer> apply(Integer t) { throw new TestException(); } }, new BiFunction<Integer, Integer, Integer>() { @Override public Integer apply(Integer t1, Integer t2) { return t1; } }).subscribe(ts); Assert.assertTrue("Not subscribed?", pp.hasSubscribers()); pp.onNext(1); Assert.assertFalse("Subscribed?", pp.hasSubscribers()); ts.assertError(TestException.class); }
@Test public void scalarReentrant() { final PublishProcessor<Flowable<Integer>> pp = PublishProcessor.create(); TestSubscriber<Integer> ts = new TestSubscriber<Integer>() { @Override public void onNext(Integer t) { super.onNext(t); if (t == 1) { pp.onNext(Flowable.just(2)); } } }; Flowable.merge(pp) .subscribe(ts); pp.onNext(Flowable.just(1)); pp.onComplete(); ts.assertResult(1, 2); }
@Test public void scalarReentrant2() { final PublishProcessor<Flowable<Integer>> pp = PublishProcessor.create(); TestSubscriber<Integer> ts = new TestSubscriber<Integer>() { @Override public void onNext(Integer t) { super.onNext(t); if (t == 1) { pp.onNext(Flowable.just(2)); } } }; Flowable.merge(pp, 2) .subscribe(ts); pp.onNext(Flowable.just(1)); pp.onComplete(); ts.assertResult(1, 2); }