Class XFlatMapTest

src/test/java/io/reactivex/rxjava3/core/XFlatMapTest.java:32–1070

import io.reactivex.rxjava3.testsupport.TestHelper;

public class XFlatMapTest extends RxJavaTest {

    @Rule
    public Retry retry = new Retry(5, 1000, true);

    static final int SLEEP_AFTER_CANCEL = 500;

    final CyclicBarrier cb = new CyclicBarrier(2);

    void sleep() throws Exception {
        cb.await();
        try {
            long before = System.currentTimeMillis();
            Thread.sleep(5000);
            throw new IllegalStateException("Was not interrupted in time?! " + (System.currentTimeMillis() - before));
        } catch (InterruptedException ex) {
            // ignored here
        }
    }

    void beforeCancelSleep(TestSubscriber<?> ts) throws Exception {
        long before = System.currentTimeMillis();
        Thread.sleep(50);
        if (System.currentTimeMillis() - before > 100) {
            ts.cancel();
            throw new IllegalStateException("Overslept?" + (System.currentTimeMillis() - before));
        }
    }

    void beforeCancelSleep(TestObserver<?> to) throws Exception {
        long before = System.currentTimeMillis();
        Thread.sleep(50);
        if (System.currentTimeMillis() - before > 100) {
            to.dispose();
            throw new IllegalStateException("Overslept?" + (System.currentTimeMillis() - before));
        }
    }

    @Test
    public void flowableFlowable() throws Exception {
        List<Throwable> errors = TestHelper.trackPluginErrors();
        try {
            TestSubscriber<Integer> ts = Flowable.just(1)
            .subscribeOn(Schedulers.io())
            .flatMap(new Function<Integer, Publisher<Integer>>() {
                @Override
                public Publisher<Integer> apply(Integer v) throws Exception {
                    sleep();
                    return Flowable.<Integer>error(new TestException());
                }
            })
            .test();

            cb.await();

            beforeCancelSleep(ts);

            ts.cancel();