/**

* Copyright 2016 Netflix, Inc.

*

* 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;

import java.util.*;

import java.util.concurrent.*;

import java.util.concurrent.atomic.*;

import org.junit.*;

import org.reactivestreams.*;

import io.reactivex.Completable.*;

import io.reactivex.NbpObservable.NbpSubscriber;

import io.reactivex.disposables.*;

import io.reactivex.exceptions.*;

import io.reactivex.functions.*;

import io.reactivex.internal.disposables.EmptyDisposable;

import io.reactivex.internal.functions.Functions;

import io.reactivex.plugins.RxJavaPlugins;

import io.reactivex.schedulers.*;

import io.reactivex.subjects.PublishSubject;

import io.reactivex.subscribers.TestSubscriber;

import io.reactivex.subscribers.nbp.NbpTestSubscriber;

/**

* 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.create(new CompletableOnSubscribe() {

@Override

public void accept(CompletableSubscriber s) {

getAndIncrement();

s.onSubscribe(EmptyDisposable.INSTANCE);

s.onComplete();

}

});

/**

* 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.create(new CompletableOnSubscribe() {

@Override

public void accept(CompletableSubscriber s) {

getAndIncrement();

s.onSubscribe(EmptyDisposable.INSTANCE);

s.onError(new TestException());

}

});

/**

* 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 = 1000)

public void complete() {

Completable c = Completable.complete();

c.await();

}

@Test(expected = NullPointerException.class)

public void concatNull() {

Completable.concat((Completable[])null);

}

@Test(timeout = 1000)

public void concatEmpty() {

Completable c = Completable.concat();

c.await();

}

@Test(timeout = 1000)

public void concatSingleSource() {

Completable c = Completable.concat(normal.completable);

c.await();

normal.assertSubscriptions(1);

}

@Test(timeout = 1000, expected = TestException.class)

public void concatSingleSourceThrows() {

Completable c = Completable.concat(error.completable);

c.await();

}

@Test(timeout = 1000)

public void concatMultipleSources() {

Completable c = Completable.concat(normal.completable, normal.completable, normal.completable);

c.await();

normal.assertSubscriptions(3);

}

@Test(timeout = 1000, expected = TestException.class)

public void concatMultipleOneThrows() {

Completable c = Completable.concat(normal.completable, error.completable, normal.completable);

c.await();

}

@Test(timeout = 1000, expected = NullPointerException.class)

public void concatMultipleOneIsNull() {

Completable c = Completable.concat(normal.completable, null);

c.await();

}

@Test(timeout = 1000)

public void concatIterableEmpty() {

Completable c = Completable.concat(Collections.<Completable>emptyList());

c.await();

}

@Test(expected = NullPointerException.class)

public void concatIterableNull() {

Completable.concat((Iterable<Completable>)null);

}

@Test(timeout = 1000, expected = NullPointerException.class)

public void concatIterableIteratorNull() {

Completable c = Completable.concat(new Iterable<Completable>() {

@Override

public Iterator<Completable> iterator() {

return null;

}

});

c.await();

}

@Test(timeout = 1000, expected = NullPointerException.class)

public void concatIterableWithNull() {

Completable c = Completable.concat(Arrays.asList(normal.completable, (Completable)null));

c.await();

}

@Test(timeout = 1000)

public void concatIterableSingle() {

Completable c = Completable.concat(Collections.singleton(normal.completable));

c.await();

normal.assertSubscriptions(1);

}

@Test(timeout = 1000)

public void concatIterableMany() {

Completable c = Completable.concat(Arrays.asList(normal.completable, normal.completable, normal.completable));

c.await();

normal.assertSubscriptions(3);

}

@Test(timeout = 1000, expected = TestException.class)

public void concatIterableOneThrows() {

Completable c = Completable.concat(Collections.singleton(error.completable));

c.await();

}

@Test(timeout = 1000, expected = TestException.class)

public void concatIterableManyOneThrows() {

Completable c = Completable.concat(Arrays.asList(normal.completable, error.completable));

c.await();

}

@Test(expected = TestException.class)

public void concatIterableIterableThrows() {

Completable c = Completable.concat(new Iterable<Completable>() {

@Override

public Iterator<Completable> iterator() {

throw new TestException();

}

});

c.await();

}

@Test(expected = TestException.class)

public void concatIterableIteratorHasNextThrows() {

Completable c = Completable.concat(new IterableIteratorHasNextThrows());

c.await();

}

@Test(expected = TestException.class)

public void concatIterableIteratorNextThrows() {

Completable c = Completable.concat(new IterableIteratorNextThrows());

c.await();

}

@Test(timeout = 1000)

public void concatObservableEmpty() {

Completable c = Completable.concat(Observable.<Completable>empty());

c.await();

}

@Test(timeout = 1000, expected = TestException.class)

public void concatObservableError() {

Completable c = Completable.concat(Observable.<Completable>error(new Supplier<Throwable>() {

@Override

public Throwable get() {

return new TestException();

}

}));

c.await();

}

@Test(timeout = 1000)

public void concatObservableSingle() {

Completable c = Completable.concat(Observable.just(normal.completable));

c.await();

normal.assertSubscriptions(1);

}

@Test(timeout = 1000, expected = TestException.class)

public void concatObservableSingleThrows() {

Completable c = Completable.concat(Observable.just(error.completable));

c.await();

}

@Test(timeout = 1000)

public void concatObservableMany() {

Completable c = Completable.concat(Observable.just(normal.completable).repeat(3));

c.await();

normal.assertSubscriptions(3);

}

@Test(timeout = 1000, expected = TestException.class)

public void concatObservableManyOneThrows() {

Completable c = Completable.concat(Observable.just(normal.completable, error.completable));

c.await();

}

@Test(timeout = 1000)

public void concatObservablePrefetch() {

final List<Long> requested = new ArrayList<Long>();

Observable<Completable> cs = Observable

.just(normal.completable)

.repeat(10)

.doOnRequest(new LongConsumer() {

@Override

public void accept(long v) {

requested.add(v);

}

});

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

c.await();

// 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 createNull() {

Completable.create(null);

}

@Test(timeout = 1000, expected = NullPointerException.class)

public void createOnSubscribeThrowsNPE() {

Completable c = Completable.create(new CompletableOnSubscribe() {

@Override

public void accept(CompletableSubscriber s) { throw new NullPointerException(); }

});

c.await();

}

@Test(timeout = 1000)

public void createOnSubscribeThrowsRuntimeException() {

try {

Completable c = Completable.create(new CompletableOnSubscribe() {

@Override

public void accept(CompletableSubscriber s) {

throw new TestException();

}

});

c.await();

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

}

}

}

@Test(timeout = 1000)

public void defer() {

Completable c = Completable.defer(new Supplier<Completable>() {

@Override

public Completable get() {

return normal.completable;

}

});

normal.assertSubscriptions(0);

c.await();

normal.assertSubscriptions(1);

}

@Test(expected = NullPointerException.class)

public void deferNull() {

Completable.defer(null);

}

@Test(timeout = 1000, expected = NullPointerException.class)

public void deferReturnsNull() {

Completable c = Completable.defer(new Supplier<Completable>() {

@Override

public Completable get() {

return null;

}

});

c.await();

}

@Test(timeout = 1000, expected = TestException.class)

public void deferFunctionThrows() {

Completable c = Completable.defer(new Supplier<Completable>() {

@Override

public Completable get() { throw new TestException(); }

});

c.await();

}

@Test(timeout = 1000, expected = TestException.class)

public void deferErrorSource() {

Completable c = Completable.defer(new Supplier<Completable>() {

@Override

public Completable get() {

return error.completable;

}

});

c.await();

}

@Test(expected = NullPointerException.class)

public void errorSupplierNull() {

Completable.error((Supplier<Throwable>)null);

}

@Test(timeout = 1000, expected = TestException.class)

public void errorSupplierNormal() {

Completable c = Completable.error(new Supplier<Throwable>() {

@Override

public Throwable get() {

return new TestException();

}

});

c.await();

}

@Test(timeout = 1000, expected = NullPointerException.class)

public void errorSupplierReturnsNull() {

Completable c = Completable.error(new Supplier<Throwable>() {

@Override

public Throwable get() {

return null;

}

});

c.await();

}

@Test(timeout = 1000, expected = TestException.class)

public void errorSupplierThrows() {

Completable c = Completable.error(new Supplier<Throwable>() {

@Override

public Throwable get() { throw new TestException(); }

});

c.await();

}

@Test(expected = NullPointerException.class)

public void errorNull() {

Completable.error((Throwable)null);

}

@Test(timeout = 1000, expected = TestException.class)

public void errorNormal() {

Completable c = Completable.error(new TestException());

c.await();

}

@Test(expected = NullPointerException.class)

public void fromCallableNull() {

Completable.fromCallable(null);

}

@Test(timeout = 1000)

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.await();

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

}

@Test(timeout = 1000, expected = TestException.class)

public void fromCallableThrows() {

Completable c = Completable.fromCallable(new Callable<Object>() {

@Override

public Object call() throws Exception { throw new TestException(); }

});

c.await();

}

@Test(expected = NullPointerException.class)

public void fromFlowableNull() {

Completable.fromFlowable(null);

}

@Test(timeout = 1000)

public void fromFlowableEmpty() {

Completable c = Completable.fromFlowable(Observable.empty());

c.await();

}

@Test(timeout = 5000)

public void fromFlowableSome() {

for (int n = 1; n < 10000; n *= 10) {

Completable c = Completable.fromFlowable(Observable.range(1, n));

c.await();

}

}

@Test(timeout = 1000, expected = TestException.class)

public void fromFlowableError() {

Completable c = Completable.fromFlowable(Observable.error(new Supplier<Throwable>() {

@Override

public Throwable get() {

return new TestException();

}

}));

c.await();

}

@Test(expected = NullPointerException.class)

public void fromNbpObservableNull() {

Completable.fromNbpObservable(null);

}

@Test(timeout = 1000)

public void fromNbpObservableEmpty() {

Completable c = Completable.fromNbpObservable(NbpObservable.empty());

c.await();

}

@Test(timeout = 5000)

public void fromNbpObservableSome() {

for (int n = 1; n < 10000; n *= 10) {

Completable c = Completable.fromNbpObservable(NbpObservable.range(1, n));

c.await();

}

}

@Test(timeout = 1000, expected = TestException.class)

public void fromNbpObservableError() {

Completable c = Completable.fromNbpObservable(NbpObservable.error(new Supplier<Throwable>() {

@Override

public Throwable get() {

return new TestException();

}

}));

c.await();

}

@Test(expected = NullPointerException.class)

public void fromRunnableNull() {

Completable.fromRunnable(null);

}

@Test(timeout = 1000)

public void fromRunnableNormal() {

final AtomicInteger calls = new AtomicInteger();

Completable c = Completable.fromRunnable(new Runnable() {

@Override

public void run() {

calls.getAndIncrement();

}

});

c.await();

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

}

@Test(timeout = 1000, expected = TestException.class)

public void fromRunnableThrows() {

Completable c = Completable.fromRunnable(new Runnable() {

@Override

public void run() { throw new TestException(); }

});

c.await();

}

@Test(expected = NullPointerException.class)

public void fromSingleNull() {

Completable.fromSingle(null);

}

@Test(timeout = 1000)

public void fromSingleNormal() {

Completable c = Completable.fromSingle(Single.just(1));

c.await();

}

@Test(timeout = 1000, expected = TestException.class)

public void fromSingleThrows() {

Completable c = Completable.fromSingle(Single.error(new Supplier<Throwable>() {

@Override

public Throwable get() {

return new TestException();

}

}));

c.await();

}

@Test(expected = NullPointerException.class)

public void mergeNull() {

Completable.merge((Completable[])null);

}

@Test(timeout = 1000)

public void mergeEmpty() {

Completable c = Completable.merge();

c.await();

}

@Test(timeout = 1000)

public void mergeSingleSource() {

Completable c = Completable.merge(normal.completable);

c.await();

normal.assertSubscriptions(1);

}

@Test(timeout = 1000, expected = TestException.class)

public void mergeSingleSourceThrows() {

Completable c = Completable.merge(error.completable);

c.await();

}

@Test(timeout = 1000)

public void mergeMultipleSources() {

Completable c = Completable.merge(normal.completable, normal.completable, normal.completable);

c.await();

normal.assertSubscriptions(3);

}

@Test(timeout = 1000, expected = TestException.class)

public void mergeMultipleOneThrows() {

Completable c = Completable.merge(normal.completable, error.completable, normal.completable);

c.await();

}

@Test(timeout = 1000, expected = NullPointerException.class)

public void mergeMultipleOneIsNull() {

Completable c = Completable.merge(normal.completable, null);

c.await();

}

@Test(timeout = 1000)

public void mergeIterableEmpty() {

Completable c = Completable.merge(Collections.<Completable>emptyList());

c.await();

}

@Test(expected = NullPointerException.class)

public void mergeIterableNull() {

Completable.merge((Iterable<Completable>)null);

}

@Test(timeout = 1000, expected = NullPointerException.class)

public void mergeIterableIteratorNull() {

Completable c = Completable.merge(new Iterable<Completable>() {

@Override

public Iterator<Completable> iterator() {

return null;

}

});

c.await();

}

@Test(timeout = 1000, expected = NullPointerException.class)

public void mergeIterableWithNull() {

Completable c = Completable.merge(Arrays.asList(normal.completable, (Completable)null));

c.await();

}

@Test(timeout = 1000)

public void mergeIterableSingle() {

Completable c = Completable.merge(Collections.singleton(normal.completable));

c.await();

normal.assertSubscriptions(1);

}

@Test(timeout = 1000)

public void mergeIterableMany() {

Completable c = Completable.merge(Arrays.asList(normal.completable, normal.completable, normal.completable));

c.await();

normal.assertSubscriptions(3);

}

@Test(timeout = 1000, expected = TestException.class)

public void mergeIterableOneThrows() {

Completable c = Completable.merge(Collections.singleton(error.completable));

c.await();

}

@Test(timeout = 1000, expected = TestException.class)

public void mergeIterableManyOneThrows() {

Completable c = Completable.merge(Arrays.asList(normal.completable, error.completable));

c.await();

}

@Test(expected = TestException.class)

public void mergeIterableIterableThrows() {

Completable c = Completable.merge(new Iterable<Completable>() {

@Override

public Iterator<Completable> iterator() {

throw new TestException();

}

});

c.await();

}

@Test(expected = TestException.class)

public void mergeIterableIteratorHasNextThrows() {

Completable c = Completable.merge(new IterableIteratorHasNextThrows());

c.await();

}

@Test(expected = TestException.class)

public void mergeIterableIteratorNextThrows() {

Completable c = Completable.merge(new IterableIteratorNextThrows());

c.await();

}

@Test(timeout = 1000)

public void mergeObservableEmpty() {

Completable c = Completable.merge(Observable.<Completable>empty());

c.await();

}

@Test(timeout = 1000, expected = TestException.class)

public void mergeObservableError() {

Completable c = Completable.merge(Observable.<Completable>error(new Supplier<Throwable>() {

@Override

public Throwable get() {

return new TestException();

}

}));

c.await();

}

@Test(timeout = 1000)

public void mergeObservableSingle() {

Completable c = Completable.merge(Observable.just(normal.completable));

c.await();

normal.assertSubscriptions(1);

}

@Test(timeout = 1000, expected = TestException.class)

public void mergeObservableSingleThrows() {

Completable c = Completable.merge(Observable.just(error.completable));

c.await();

}

@Test(timeout = 1000)

public void mergeObservableMany() {

Completable c = Completable.merge(Observable.just(normal.completable).repeat(3));

c.await();

normal.assertSubscriptions(3);

}

@Test(timeout = 1000, expected = TestException.class)

public void mergeObservableManyOneThrows() {

Completable c = Completable.merge(Observable.just(normal.completable, error.completable));

c.await();

}

@Test(timeout = 1000)

public void mergeObservableMaxConcurrent() {

final List<Long> requested = new ArrayList<Long>();

Observable<Completable> cs = Observable

.just(normal.completable)

.repeat(10)

.doOnRequest(new LongConsumer() {

@Override

public void accept(long v) {

requested.add(v);

}

});

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

c.await();

// 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.mergeDelayError((Completable[])null);

}

@Test(timeout = 1000)

public void mergeDelayErrorEmpty() {

Completable c = Completable.mergeDelayError();

c.await();

}

@Test(timeout = 1000)

public void mergeDelayErrorSingleSource() {

Completable c = Completable.mergeDelayError(normal.completable);

c.await();

normal.assertSubscriptions(1);

}

@Test(timeout = 1000, expected = TestException.class)

public void mergeDelayErrorSingleSourceThrows() {

Completable c = Completable.mergeDelayError(error.completable);

c.await();

}

@Test(timeout = 1000)

public void mergeDelayErrorMultipleSources() {

Completable c = Completable.mergeDelayError(normal.completable, normal.completable, normal.completable);

c.await();

normal.assertSubscriptions(3);

}

@Test(timeout = 1000)

public void mergeDelayErrorMultipleOneThrows() {

Completable c = Completable.mergeDelayError(normal.completable, error.completable, normal.completable);

try {

c.await();

} catch (TestException ex) {

normal.assertSubscriptions(2);

}

}

@Test(timeout = 1000, expected = NullPointerException.class)

public void mergeDelayErrorMultipleOneIsNull() {

Completable c = Completable.mergeDelayError(normal.completable, null);

c.await();

}

@Test(timeout = 1000)

public void mergeDelayErrorIterableEmpty() {

Completable c = Completable.mergeDelayError(Collections.<Completable>emptyList());

c.await();

}

@Test(expected = NullPointerException.class)

public void mergeDelayErrorIterableNull() {

Completable.mergeDelayError((Iterable<Completable>)null);

}

@Test(timeout = 1000, expected = NullPointerException.class)

public void mergeDelayErrorIterableIteratorNull() {

Completable c = Completable.mergeDelayError(new Iterable<Completable>() {

@Override

public Iterator<Completable> iterator() {

return null;

}

});

c.await();

}

@Test(timeout = 1000, expected = NullPointerException.class)

public void mergeDelayErrorIterableWithNull() {

Completable c = Completable.mergeDelayError(Arrays.asList(normal.completable, (Completable)null));

c.await();

}

@Test(timeout = 1000)

public void mergeDelayErrorIterableSingle() {

Completable c = Completable.mergeDelayError(Collections.singleton(normal.completable));

c.await();

normal.assertSubscriptions(1);

}

@Test(timeout = 1000)

public void mergeDelayErrorIterableMany() {

Completable c = Completable.mergeDelayError(Arrays.asList(normal.completable, normal.completable, normal.completable));

c.await();

normal.assertSubscriptions(3);

}

@Test(timeout = 1000, expected = TestException.class)

public void mergeDelayErrorIterableOneThrows() {

Completable c = Completable.mergeDelayError(Collections.singleton(error.completable));

c.await();

}

@Test(timeout = 1000)

public void mergeDelayErrorIterableManyOneThrows() {

Completable c = Completable.mergeDelayError(Arrays.asList(normal.completable, error.completable, normal.completable));

try {

c.await();

} catch (TestException ex) {

normal.assertSubscriptions(2);

}

}

@Test(expected = TestException.class)

public void mergeDelayErrorIterableIterableThrows() {

Completable c = Completable.mergeDelayError(new Iterable<Completable>() {

@Override

public Iterator<Completable> iterator() {