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