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