CompletableTest.java

/**
 * Copyright (c) 2016-present, RxJava Contributors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
 * compliance with the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License is
 * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
 * the License for the specific language governing permissions and limitations under the License.
 */

package io.reactivex.completable;

import static org.junit.Assert.*;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

import org.junit.*;
import org.reactivestreams.*;

import io.reactivex.*;
import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.*;
import io.reactivex.exceptions.*;
import io.reactivex.functions.*;
import io.reactivex.internal.disposables.*;
import io.reactivex.internal.functions.Functions;
import io.reactivex.internal.subscriptions.BooleanSubscription;
import io.reactivex.internal.util.ExceptionHelper;
import io.reactivex.observers.TestObserver;
import io.reactivex.plugins.RxJavaPlugins;
import io.reactivex.processors.PublishProcessor;
import io.reactivex.schedulers.*;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subscribers.TestSubscriber;

/**
 * Test Completable methods and operators.
 */
public class CompletableTest {
    /**
     * Iterable that returns an Iterator that throws in its hasNext method.
     */
    static final class IterableIteratorNextThrows implements Iterable<Completable> {
        @Override
        public Iterator<Completable> iterator() {
            return new Iterator<Completable>() {
                @Override
                public boolean hasNext() {
                    return true;
                }

                @Override
                public Completable next() {
                    throw new TestException();
                }

                @Override
                public void remove() {
                    // TODO Auto-generated method stub

                }
            };
        }
    }

    /**
     * Iterable that returns an Iterator that throws in its next method.
     */
    static final class IterableIteratorHasNextThrows implements Iterable<Completable> {
        @Override
        public Iterator<Completable> iterator() {
            return new Iterator<Completable>() {
                @Override
                public boolean hasNext() {
                    throw new TestException();
                }

                @Override
                public Completable next() {
                    return null;
                }

                @Override
                public void remove() {
                    // TODO Auto-generated method stub

                }
            };
        }
    }

    /**
     * A class containing a completable instance and counts the number of subscribers.
     */
    static final class NormalCompletable extends AtomicInteger {

        private static final long serialVersionUID = 7192337844700923752L;

        public final Completable completable = Completable.unsafeCreate(new CompletableSource() {
            @Override
            public void subscribe(CompletableObserver observer) {
                getAndIncrement();
                EmptyDisposable.complete(observer);
            }
        });

        /**
         * Asserts the given number of subscriptions happened.
         * @param n the expected number of subscriptions
         */
        public void assertSubscriptions(int n) {
            Assert.assertEquals(n, get());
        }
    }

    /**
     * A class containing a completable instance that emits a TestException and counts
     * the number of subscribers.
     */
    static final class ErrorCompletable extends AtomicInteger {

        private static final long serialVersionUID = 7192337844700923752L;

        public final Completable completable = Completable.unsafeCreate(new CompletableSource() {
            @Override
            public void subscribe(CompletableObserver observer) {
                getAndIncrement();
                EmptyDisposable.error(new TestException(), observer);
            }
        });

        /**
         * Asserts the given number of subscriptions happened.
         * @param n the expected number of subscriptions
         */
        public void assertSubscriptions(int n) {
            Assert.assertEquals(n, get());
        }
    }

    /** A normal Completable object. */
    final NormalCompletable normal = new NormalCompletable();

    /** An error Completable object. */
    final ErrorCompletable error = new ErrorCompletable();

    @Test(timeout = 5000)
    public void complete() {
        Completable c = Completable.complete();

        c.blockingAwait();
    }

    @Test(expected = NullPointerException.class)
    public void concatNull() {
        Completable.concatArray((Completable[])null);
    }

    @Test(timeout = 5000)
    public void concatEmpty() {
        Completable c = Completable.concatArray();

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void concatSingleSource() {
        Completable c = Completable.concatArray(normal.completable);

        c.blockingAwait();

        normal.assertSubscriptions(1);
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void concatSingleSourceThrows() {
        Completable c = Completable.concatArray(error.completable);

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void concatMultipleSources() {
        Completable c = Completable.concatArray(normal.completable, normal.completable, normal.completable);

        c.blockingAwait();

        normal.assertSubscriptions(3);
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void concatMultipleOneThrows() {
        Completable c = Completable.concatArray(normal.completable, error.completable, normal.completable);

        c.blockingAwait();
    }

    @Test(timeout = 5000, expected = NullPointerException.class)
    public void concatMultipleOneIsNull() {
        Completable c = Completable.concatArray(normal.completable, null);

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void concatIterableEmpty() {
        Completable c = Completable.concat(Collections.<Completable>emptyList());

        c.blockingAwait();
    }

    @Test(expected = NullPointerException.class)
    public void concatIterableNull() {
        Completable.concat((Iterable<Completable>)null);
    }

    @Test(timeout = 5000, expected = NullPointerException.class)
    public void concatIterableIteratorNull() {
        Completable c = Completable.concat(new Iterable<Completable>() {
            @Override
            public Iterator<Completable> iterator() {
                return null;
            }
        });

        c.blockingAwait();
    }

    @Test(timeout = 5000, expected = NullPointerException.class)
    public void concatIterableWithNull() {
        Completable c = Completable.concat(Arrays.asList(normal.completable, (Completable)null));

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void concatIterableSingle() {
        Completable c = Completable.concat(Collections.singleton(normal.completable));

        c.blockingAwait();

        normal.assertSubscriptions(1);
    }

    @Test(timeout = 5000)
    public void concatIterableMany() {
        Completable c = Completable.concat(Arrays.asList(normal.completable, normal.completable, normal.completable));

        c.blockingAwait();

        normal.assertSubscriptions(3);
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void concatIterableOneThrows() {
        Completable c = Completable.concat(Collections.singleton(error.completable));

        c.blockingAwait();
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void concatIterableManyOneThrows() {
        Completable c = Completable.concat(Arrays.asList(normal.completable, error.completable));

        c.blockingAwait();
    }

    @Test(expected = TestException.class)
    public void concatIterableIterableThrows() {
        Completable c = Completable.concat(new Iterable<Completable>() {
            @Override
            public Iterator<Completable> iterator() {
                throw new TestException();
            }
        });

        c.blockingAwait();
    }

    @Test(expected = TestException.class)
    public void concatIterableIteratorHasNextThrows() {
        Completable c = Completable.concat(new IterableIteratorHasNextThrows());

        c.blockingAwait();
    }

    @Test(expected = TestException.class)
    public void concatIterableIteratorNextThrows() {
        Completable c = Completable.concat(new IterableIteratorNextThrows());

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void concatObservableEmpty() {
        Completable c = Completable.concat(Flowable.<Completable>empty());

        c.blockingAwait();
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void concatObservableError() {
        Completable c = Completable.concat(Flowable.<Completable>error(new Callable<Throwable>() {
            @Override
            public Throwable call() {
                return new TestException();
            }
        }));

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void concatObservableSingle() {
        Completable c = Completable.concat(Flowable.just(normal.completable));

        c.blockingAwait();

        normal.assertSubscriptions(1);
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void concatObservableSingleThrows() {
        Completable c = Completable.concat(Flowable.just(error.completable));

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void concatObservableMany() {
        Completable c = Completable.concat(Flowable.just(normal.completable).repeat(3));

        c.blockingAwait();

        normal.assertSubscriptions(3);
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void concatObservableManyOneThrows() {
        Completable c = Completable.concat(Flowable.just(normal.completable, error.completable));

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void concatObservablePrefetch() {
        final List<Long> requested = new ArrayList<Long>();
        Flowable<Completable> cs = Flowable
                .just(normal.completable)
                .repeat(10)
                .doOnRequest(new LongConsumer() {
                    @Override
                    public void accept(long v) {
                        requested.add(v);
                    }
                });

        Completable c = Completable.concat(cs, 5);

        c.blockingAwait();

        Assert.assertEquals(Arrays.asList(5L, 4L, 4L), requested);
    }

    @Test(expected = NullPointerException.class)
    public void createNull() {
        Completable.unsafeCreate(null);
    }

    @Test(timeout = 5000, expected = NullPointerException.class)
    public void createOnSubscribeThrowsNPE() {
        Completable c = Completable.unsafeCreate(new CompletableSource() {
            @Override
            public void subscribe(CompletableObserver observer) { throw new NullPointerException(); }
        });

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void createOnSubscribeThrowsRuntimeException() {
        List<Throwable> errors = TestHelper.trackPluginErrors();
        try {
            Completable c = Completable.unsafeCreate(new CompletableSource() {
                @Override
                public void subscribe(CompletableObserver observer) {
                    throw new TestException();
                }
            });

            c.blockingAwait();

            Assert.fail("Did not throw exception");
        } catch (NullPointerException ex) {
            if (!(ex.getCause() instanceof TestException)) {
                ex.printStackTrace();
                Assert.fail("Did not wrap the TestException but it returned: " + ex);
            }

            TestHelper.assertUndeliverable(errors, 0, TestException.class);
        } finally {
            RxJavaPlugins.reset();
        }
    }

    @Test(timeout = 5000)
    public void defer() {
        Completable c = Completable.defer(new Callable<Completable>() {
            @Override
            public Completable call() {
                return normal.completable;
            }
        });

        normal.assertSubscriptions(0);

        c.blockingAwait();

        normal.assertSubscriptions(1);
    }

    @Test(expected = NullPointerException.class)
    public void deferNull() {
        Completable.defer(null);
    }

    @Test(timeout = 5000, expected = NullPointerException.class)
    public void deferReturnsNull() {
        Completable c = Completable.defer(new Callable<Completable>() {
            @Override
            public Completable call() {
                return null;
            }
        });

        c.blockingAwait();
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void deferFunctionThrows() {
        Completable c = Completable.defer(new Callable<Completable>() {
            @Override
            public Completable call() { throw new TestException(); }
        });

        c.blockingAwait();
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void deferErrorSource() {
        Completable c = Completable.defer(new Callable<Completable>() {
            @Override
            public Completable call() {
                return error.completable;
            }
        });

        c.blockingAwait();
    }

    @Test(expected = NullPointerException.class)
    public void errorSupplierNull() {
        Completable.error((Callable<Throwable>)null);
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void errorSupplierNormal() {
        Completable c = Completable.error(new Callable<Throwable>() {
            @Override
            public Throwable call() {
                return new TestException();
            }
        });

        c.blockingAwait();
    }

    @Test(timeout = 5000, expected = NullPointerException.class)
    public void errorSupplierReturnsNull() {
        Completable c = Completable.error(new Callable<Throwable>() {
            @Override
            public Throwable call() {
                return null;
            }
        });

        c.blockingAwait();
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void errorSupplierThrows() {
        Completable c = Completable.error(new Callable<Throwable>() {
            @Override
            public Throwable call() { throw new TestException(); }
        });

        c.blockingAwait();
    }

    @Test(expected = NullPointerException.class)
    public void errorNull() {
        Completable.error((Throwable)null);
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void errorNormal() {
        Completable c = Completable.error(new TestException());

        c.blockingAwait();
    }

    @Test(expected = NullPointerException.class)
    public void fromCallableNull() {
        Completable.fromCallable(null);
    }

    @Test(timeout = 5000)
    public void fromCallableNormal() {
        final AtomicInteger calls = new AtomicInteger();

        Completable c = Completable.fromCallable(new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                return calls.getAndIncrement();
            }
        });

        c.blockingAwait();

        Assert.assertEquals(1, calls.get());
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void fromCallableThrows() {
        Completable c = Completable.fromCallable(new Callable<Object>() {
            @Override
            public Object call() throws Exception { throw new TestException(); }
        });

        c.blockingAwait();
    }

    @Test(expected = NullPointerException.class)
    public void fromFlowableNull() {
        Completable.fromPublisher(null);
    }

    @Test(timeout = 5000)
    public void fromFlowableEmpty() {
        Completable c = Completable.fromPublisher(Flowable.empty());

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void fromFlowableSome() {
        for (int n = 1; n < 10000; n *= 10) {
            Completable c = Completable.fromPublisher(Flowable.range(1, n));

            c.blockingAwait();
        }
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void fromFlowableError() {
        Completable c = Completable.fromPublisher(Flowable.error(new Callable<Throwable>() {
            @Override
            public Throwable call() {
                return new TestException();
            }
        }));

        c.blockingAwait();
    }

    @Test(expected = NullPointerException.class)
    public void fromObservableNull() {
        Completable.fromObservable(null);
    }

    @Test(timeout = 5000)
    public void fromObservableEmpty() {
        Completable c = Completable.fromObservable(Observable.empty());

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void fromObservableSome() {
        for (int n = 1; n < 10000; n *= 10) {
            Completable c = Completable.fromObservable(Observable.range(1, n));

            c.blockingAwait();
        }
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void fromObservableError() {
        Completable c = Completable.fromObservable(Observable.error(new Callable<Throwable>() {
            @Override
            public Throwable call() {
                return new TestException();
            }
        }));

        c.blockingAwait();
    }

    @Test(expected = NullPointerException.class)
    public void fromActionNull() {
        Completable.fromAction(null);
    }

    @Test(timeout = 5000)
    public void fromActionNormal() {
        final AtomicInteger calls = new AtomicInteger();

        Completable c = Completable.fromAction(new Action() {
            @Override
            public void run() {
                calls.getAndIncrement();
            }
        });

        c.blockingAwait();

        Assert.assertEquals(1, calls.get());
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void fromActionThrows() {
        Completable c = Completable.fromAction(new Action() {
            @Override
            public void run() { throw new TestException(); }
        });

        c.blockingAwait();
    }

    @Test(expected = NullPointerException.class)
    public void fromSingleNull() {
        Completable.fromSingle(null);
    }

    @Test(timeout = 5000)
    public void fromSingleNormal() {
        Completable c = Completable.fromSingle(Single.just(1));

        c.blockingAwait();
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void fromSingleThrows() {
        Completable c = Completable.fromSingle(Single.error(new Callable<Throwable>() {
            @Override
            public Throwable call() {
                return new TestException();
            }
        }));

        c.blockingAwait();
    }

    @Test(expected = NullPointerException.class)
    public void mergeNull() {
        Completable.mergeArray((Completable[])null);
    }

    @Test(timeout = 5000)
    public void mergeEmpty() {
        Completable c = Completable.mergeArray();

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void mergeSingleSource() {
        Completable c = Completable.mergeArray(normal.completable);

        c.blockingAwait();

        normal.assertSubscriptions(1);
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void mergeSingleSourceThrows() {
        Completable c = Completable.mergeArray(error.completable);

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void mergeMultipleSources() {
        Completable c = Completable.mergeArray(normal.completable, normal.completable, normal.completable);

        c.blockingAwait();

        normal.assertSubscriptions(3);
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void mergeMultipleOneThrows() {
        Completable c = Completable.mergeArray(normal.completable, error.completable, normal.completable);

        c.blockingAwait();
    }

    @Test(timeout = 5000, expected = NullPointerException.class)
    public void mergeMultipleOneIsNull() {
        Completable c = Completable.mergeArray(normal.completable, null);

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void mergeIterableEmpty() {
        Completable c = Completable.merge(Collections.<Completable>emptyList());

        c.blockingAwait();
    }

    @Test(expected = NullPointerException.class)
    public void mergeIterableNull() {
        Completable.merge((Iterable<Completable>)null);
    }

    @Test(timeout = 5000, expected = NullPointerException.class)
    public void mergeIterableIteratorNull() {
        Completable c = Completable.merge(new Iterable<Completable>() {
            @Override
            public Iterator<Completable> iterator() {
                return null;
            }
        });

        c.blockingAwait();
    }

    @Test(timeout = 5000, expected = NullPointerException.class)
    public void mergeIterableWithNull() {
        Completable c = Completable.merge(Arrays.asList(normal.completable, (Completable)null));

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void mergeIterableSingle() {
        Completable c = Completable.merge(Collections.singleton(normal.completable));

        c.blockingAwait();

        normal.assertSubscriptions(1);
    }

    @Test(timeout = 5000)
    public void mergeIterableMany() {
        Completable c = Completable.merge(Arrays.asList(normal.completable, normal.completable, normal.completable));

        c.blockingAwait();

        normal.assertSubscriptions(3);
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void mergeIterableOneThrows() {
        Completable c = Completable.merge(Collections.singleton(error.completable));

        c.blockingAwait();
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void mergeIterableManyOneThrows() {
        Completable c = Completable.merge(Arrays.asList(normal.completable, error.completable));

        c.blockingAwait();
    }

    @Test(expected = TestException.class)
    public void mergeIterableIterableThrows() {
        Completable c = Completable.merge(new Iterable<Completable>() {
            @Override
            public Iterator<Completable> iterator() {
                throw new TestException();
            }
        });

        c.blockingAwait();
    }

    @Test(expected = TestException.class)
    public void mergeIterableIteratorHasNextThrows() {
        Completable c = Completable.merge(new IterableIteratorHasNextThrows());

        c.blockingAwait();
    }

    @Test(expected = TestException.class)
    public void mergeIterableIteratorNextThrows() {
        Completable c = Completable.merge(new IterableIteratorNextThrows());

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void mergeObservableEmpty() {
        Completable c = Completable.merge(Flowable.<Completable>empty());

        c.blockingAwait();
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void mergeObservableError() {
        Completable c = Completable.merge(Flowable.<Completable>error(new Callable<Throwable>() {
            @Override
            public Throwable call() {
                return new TestException();
            }
        }));

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void mergeObservableSingle() {
        Completable c = Completable.merge(Flowable.just(normal.completable));

        c.blockingAwait();

        normal.assertSubscriptions(1);
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void mergeObservableSingleThrows() {
        Completable c = Completable.merge(Flowable.just(error.completable));

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void mergeObservableMany() {
        Completable c = Completable.merge(Flowable.just(normal.completable).repeat(3));

        c.blockingAwait();

        normal.assertSubscriptions(3);
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void mergeObservableManyOneThrows() {
        Completable c = Completable.merge(Flowable.just(normal.completable, error.completable));

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void mergeObservableMaxConcurrent() {
        final List<Long> requested = new ArrayList<Long>();
        Flowable<Completable> cs = Flowable
                .just(normal.completable)
                .repeat(10)
                .doOnRequest(new LongConsumer() {
                    @Override
                    public void accept(long v) {
                        requested.add(v);
                    }
                });

        Completable c = Completable.merge(cs, 5);

        c.blockingAwait();

        // FIXME this request pattern looks odd because all 10 completions trigger 1 requests
        Assert.assertEquals(Arrays.asList(5L, 1L, 1L, 1L, 1L, 1L, 1L, 1L, 1L, 1L, 1L), requested);
    }

    @Test(expected = NullPointerException.class)
    public void mergeDelayErrorNull() {
        Completable.mergeArrayDelayError((Completable[])null);
    }

    @Test(timeout = 5000)
    public void mergeDelayErrorEmpty() {
        Completable c = Completable.mergeArrayDelayError();

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void mergeDelayErrorSingleSource() {
        Completable c = Completable.mergeArrayDelayError(normal.completable);

        c.blockingAwait();

        normal.assertSubscriptions(1);
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void mergeDelayErrorSingleSourceThrows() {
        Completable c = Completable.mergeArrayDelayError(error.completable);

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void mergeDelayErrorMultipleSources() {
        Completable c = Completable.mergeArrayDelayError(normal.completable, normal.completable, normal.completable);

        c.blockingAwait();

        normal.assertSubscriptions(3);
    }

    @Test(timeout = 5000)
    public void mergeDelayErrorMultipleOneThrows() {
        Completable c = Completable.mergeArrayDelayError(normal.completable, error.completable, normal.completable);

        try {
            c.blockingAwait();
        } catch (TestException ex) {
            normal.assertSubscriptions(2);
        }
    }

    @Test(timeout = 5000, expected = NullPointerException.class)
    public void mergeDelayErrorMultipleOneIsNull() {
        Completable c = Completable.mergeArrayDelayError(normal.completable, null);

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void mergeDelayErrorIterableEmpty() {
        Completable c = Completable.mergeDelayError(Collections.<Completable>emptyList());

        c.blockingAwait();
    }

    @Test(expected = NullPointerException.class)
    public void mergeDelayErrorIterableNull() {
        Completable.mergeDelayError((Iterable<Completable>)null);
    }

    @Test(timeout = 5000, expected = NullPointerException.class)
    public void mergeDelayErrorIterableIteratorNull() {
        Completable c = Completable.mergeDelayError(new Iterable<Completable>() {
            @Override
            public Iterator<Completable> iterator() {
                return null;
            }
        });

        c.blockingAwait();
    }

    @Test(timeout = 5000, expected = NullPointerException.class)
    public void mergeDelayErrorIterableWithNull() {
        Completable c = Completable.mergeDelayError(Arrays.asList(normal.completable, (Completable)null));

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void mergeDelayErrorIterableSingle() {
        Completable c = Completable.mergeDelayError(Collections.singleton(normal.completable));

        c.blockingAwait();

        normal.assertSubscriptions(1);
    }

    @Test(timeout = 5000)
    public void mergeDelayErrorIterableMany() {
        Completable c = Completable.mergeDelayError(Arrays.asList(normal.completable, normal.completable, normal.completable));

        c.blockingAwait();

        normal.assertSubscriptions(3);
    }

    @Test(timeout = 5000, expected = TestException.class)
    public void mergeDelayErrorIterableOneThrows() {
        Completable c = Completable.mergeDelayError(Collections.singleton(error.completable));

        c.blockingAwait();
    }

    @Test(timeout = 5000)
    public void mergeDelayErrorIterableManyOneThrows() {
        Completable c = Completable.mergeDelayError(Arrays.asList(normal.completable, error.completable, normal.completable));

        try {
            c.blockingAwait();