@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()));
}