/**

* Copyright 2014 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 rx;

import static org.junit.Assert.*;

import static org.mockito.Matchers.*;

import static org.mockito.Mockito.*;

import java.util.*;

import java.util.concurrent.*;

import java.util.concurrent.atomic.*;

import org.junit.*;

import rx.Completable.*;

import rx.Observable.OnSubscribe;

import rx.exceptions.*;

import rx.functions.*;

import rx.observers.TestSubscriber;

import rx.plugins.RxJavaHooks;

import rx.schedulers.*;

import rx.subjects.PublishSubject;

import rx.subscriptions.*;

/**

* 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() {

throw new UnsupportedOperationException();

}

};

}

}

/**

* 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() {

throw new UnsupportedOperationException();

}

};

}

}

/**

* 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 call(CompletableSubscriber s) {

getAndIncrement();

s.onSubscribe(Subscriptions.unsubscribed());

s.onCompleted();

}

});

/**

* 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 call(CompletableSubscriber s) {

getAndIncrement();

s.onSubscribe(Subscriptions.unsubscribed());

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 = 5000)

public void complete() {

Completable c = Completable.complete();

c.await();

}

@Test(expected = NullPointerException.class)

public void concatNull() {

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

}

@Test(timeout = 5000)

public void concatEmpty() {

Completable c = Completable.concat();

c.await();

}

@Test(timeout = 5000)

public void concatSingleSource() {

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

c.await();

normal.assertSubscriptions(1);

}

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

public void concatSingleSourceThrows() {

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

c.await();

}

@Test(timeout = 5000)

public void concatMultipleSources() {

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

c.await();

normal.assertSubscriptions(3);

}

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

public void concatMultipleOneThrows() {

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

c.await();

}

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

public void concatMultipleOneIsNull() {

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

c.await();

}

@Test(timeout = 5000)

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 = 5000, expected = NullPointerException.class)

public void concatIterableIteratorNull() {

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

@Override

public Iterator<Completable> iterator() {

return null;

}

});

c.await();

}

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

public void concatIterableWithNull() {

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

c.await();

}

@Test(timeout = 5000)

public void concatIterableSingle() {

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

c.await();

normal.assertSubscriptions(1);

}

@Test(timeout = 5000)

public void concatIterableMany() {

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

c.await();

normal.assertSubscriptions(3);

}

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

public void concatIterableOneThrows() {

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

c.await();

}

@Test(timeout = 5000, 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 = 5000)

public void concatObservableEmpty() {

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

c.await();

}

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

public void concatObservableError() {

Completable c = Completable.concat(Observable.<Completable>error(new TestException()));

c.await();

}

@Test(timeout = 5000)

public void concatObservableSingle() {

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

c.await();

normal.assertSubscriptions(1);

}

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

public void concatObservableSingleThrows() {

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

c.await();

}

@Test(timeout = 5000)

public void concatObservableMany() {

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

c.await();

normal.assertSubscriptions(3);

}

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

public void concatObservableManyOneThrows() {

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

c.await();

}

@Test(timeout = 5000)

public void concatObservablePrefetch() {

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

Observable<Completable> cs = Observable

.just(normal.completable)

.repeat(10)

.doOnRequest(new Action1<Long>() {

@Override

public void call(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

public void andThen() {

TestSubscriber<String> ts = new TestSubscriber<String>(0);

Completable.complete().andThen(Observable.just("foo")).subscribe(ts);

ts.requestMore(1);

ts.assertValue("foo");

ts.assertCompleted();

ts.assertNoErrors();

}

@Test

public void andThenNever() {

TestSubscriber<String> ts = new TestSubscriber<String>(0);

Completable.never().andThen(Observable.just("foo")).subscribe(ts);

ts.requestMore(1);

ts.assertNoValues();

ts.assertNoTerminalEvent();

}

@Test

public void andThenError() {

TestSubscriber<String> ts = new TestSubscriber<String>(0);

final AtomicBoolean hasRun = new AtomicBoolean(false);

final Exception e = new Exception();

Completable.create(new CompletableOnSubscribe() {

@Override

public void call(CompletableSubscriber cs) {

cs.onError(e);

}

})

.andThen(Observable.<String>create(new OnSubscribe<String>() {

@Override

public void call(Subscriber<? super String> s) {

hasRun.set(true);

s.onNext("foo");

s.onCompleted();

}

}))

.subscribe(ts);

ts.assertNoValues();

ts.assertError(e);

Assert.assertFalse("Should not have subscribed to observable when completable errors", hasRun.get());

}

@Test

public void andThenSubscribeOn() {

TestSubscriber<String> ts = new TestSubscriber<String>(0);

TestScheduler scheduler = new TestScheduler();

Completable.complete().andThen(Observable.just("foo").delay(1, TimeUnit.SECONDS, scheduler)).subscribe(ts);

ts.requestMore(1);

ts.assertNoValues();

ts.assertNoTerminalEvent();

scheduler.advanceTimeBy(1, TimeUnit.SECONDS);

ts.assertValue("foo");

ts.assertCompleted();

ts.assertNoErrors();

}

@Test

public void andThenSingle() {

TestSubscriber<String> ts = new TestSubscriber<String>(0);

Completable.complete().andThen(Single.just("foo")).subscribe(ts);

ts.requestMore(1);

ts.assertValue("foo");

ts.assertCompleted();

ts.assertNoErrors();

ts.assertUnsubscribed();

}

@Test

public void andThenSingleNever() {

TestSubscriber<String> ts = new TestSubscriber<String>(0);

Completable.never().andThen(Single.just("foo")).subscribe(ts);

ts.requestMore(1);

ts.assertNoValues();

ts.assertNoTerminalEvent();

}

@Test

public void andThenSingleError() {

TestSubscriber<String> ts = new TestSubscriber<String>(0);

final AtomicBoolean hasRun = new AtomicBoolean(false);

final Exception e = new Exception();

Completable.error(e)

.andThen(Single.<String>create(new Single.OnSubscribe<String>() {

@Override

public void call(SingleSubscriber<? super String> s) {

hasRun.set(true);

s.onSuccess("foo");

}

}))

.subscribe(ts);

ts.assertNoValues();

ts.assertError(e);

ts.assertUnsubscribed();

Assert.assertFalse("Should not have subscribed to single when completable errors", hasRun.get());

}

@Test

public void andThenSingleSubscribeOn() {

TestSubscriber<String> ts = new TestSubscriber<String>(0);

TestScheduler scheduler = new TestScheduler();

Completable.complete().andThen(Single.just("foo").delay(1, TimeUnit.SECONDS, scheduler)).subscribe(ts);

ts.requestMore(1);

ts.assertNoValues();

ts.assertNoTerminalEvent();

scheduler.advanceTimeBy(1, TimeUnit.SECONDS);

ts.assertValue("foo");

ts.assertCompleted();

ts.assertNoErrors();

ts.assertUnsubscribed();

}

@Test(expected = NullPointerException.class)

public void createNull() {

Completable.create(null);

}

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

public void createOnSubscribeThrowsNPE() {

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

@Override

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

});

c.await();

}

@Test(timeout = 5000)

public void createOnSubscribeThrowsRuntimeException() {

try {

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

@Override

public void call(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 = 5000)

public void defer() {

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

@Override

public Completable call() {

return normal.completable;

}

});

normal.assertSubscriptions(0);

c.await();

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 Func0<Completable>() {

@Override

public Completable call() {

return null;

}

});

c.await();

}

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

public void deferFunctionThrows() {

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

@Override

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

});

c.await();

}

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

public void deferErrorSource() {

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

@Override

public Completable call() {

return error.completable;

}

});

c.await();

}

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

}

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

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

}

@Test(expected = NullPointerException.class)

public void fromObservableNull() {

Completable.fromObservable(null);

}

@Test(timeout = 5000)

public void fromObservableEmpty() {

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

c.await();

}

@Test(timeout = 5000)

public void fromObservableSome() {

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

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

c.await();

}

}

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

public void fromObservableError() {

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

c.await();

}

@Test(expected = NullPointerException.class)

public void fromFutureNull() {

Completable.fromFuture(null);

}

@Test(timeout = 5000)

public void fromFutureNormal() {

ExecutorService exec = Executors.newSingleThreadExecutor();

try {

Completable c = Completable.fromFuture(exec.submit(new Runnable() {

@Override

public void run() {

// no action

}

}));

c.await();

} finally {

exec.shutdown();

}

}

@Test(timeout = 5000)

public void fromFutureThrows() {

ExecutorService exec = Executors.newSingleThreadExecutor();

Completable c = Completable.fromFuture(exec.submit(new Runnable() {

@Override

public void run() {

throw new TestException();

}

}));

try {

c.await();

Assert.fail("Failed to throw Exception");

} catch (RuntimeException ex) {

if (!((ex.getCause() instanceof ExecutionException) && (ex.getCause().getCause() instanceof TestException))) {

ex.printStackTrace();

Assert.fail("Wrong exception received");

}

} finally {

exec.shutdown();

}

}

@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 Action0() {

@Override

public void call() {

calls.getAndIncrement();

}

});

c.await();

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

}

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

public void fromActionThrows() {

Completable c = Completable.fromAction(new Action0() {

@Override

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

});

c.await();

}

@Test(expected = NullPointerException.class)

public void fromSingleNull() {

Completable.fromSingle(null);

}

@Test(timeout = 5000)

public void fromSingleNormal() {

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

c.await();

}

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

public void fromSingleThrows() {

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

c.await();

}

@Test(expected = NullPointerException.class)

public void mergeNull() {

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

}

@Test(timeout = 5000)

public void mergeEmpty() {

Completable c = Completable.merge();

c.await();

}

@Test(timeout = 5000)

public void mergeSingleSource() {

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

c.await();

normal.assertSubscriptions(1);

}

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

public void mergeSingleSourceThrows() {

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

c.await();

}

@Test(timeout = 5000)

public void mergeMultipleSources() {

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

c.await();

normal.assertSubscriptions(3);

}

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

public void mergeMultipleOneThrows() {

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

c.await();

}

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

public void mergeMultipleOneIsNull() {

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

c.await();

}

@Test(timeout = 5000)

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 = 5000, expected = NullPointerException.class)

public void mergeIterableIteratorNull() {

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

@Override

public Iterator<Completable> iterator() {

return null;

}

});

c.await();

}

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

public void mergeIterableWithNull() {

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

c.await();

}

@Test(timeout = 5000)

public void mergeIterableSingle() {

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

c.await();

normal.assertSubscriptions(1);

}

@Test(timeout = 5000)

public void mergeIterableMany() {

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

c.await();

normal.assertSubscriptions(3);

}

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

public void mergeIterableOneThrows() {

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

c.await();

}

@Test(timeout = 5000, 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 = 5000)

public void mergeObservableEmpty() {

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

c.await();

}

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

public void mergeObservableError() {

Completable c = Completable.merge(Observable.<Completable>error(new TestException()));

c.await();

}

@Test(timeout = 5000)

public void mergeObservableSingle() {

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

c.await();

normal.assertSubscriptions(1);

}

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

public void mergeObservableSingleThrows() {

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

c.await();

}

@Test(timeout = 5000)

public void mergeObservableMany() {

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

c.await();

normal.assertSubscriptions(3);

}

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

public void mergeObservableManyOneThrows() {

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

c.await();

}

@Test(timeout = 5000)

public void mergeObservableMaxConcurrent() {

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

Observable<Completable> cs = Observable

.just(normal.completable)

.repeat(10)

.doOnRequest(new Action1<Long>() {

@Override

public void call(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 = 5000)

public void mergeDelayErrorEmpty() {

Completable c = Completable.mergeDelayError();

c.await();

}

@Test(timeout = 5000)

public void mergeDelayErrorSingleSource() {

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

c.await();

normal.assertSubscriptions(1);

}

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

public void mergeDelayErrorSingleSourceThrows() {

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

c.await();

}

@Test(timeout = 5000)

public void mergeDelayErrorMultipleSources() {

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

c.await();

normal.assertSubscriptions(3);

}

@Test(timeout = 5000)

public void mergeDelayErrorMultipleOneThrows() {

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

try {

c.await();

} catch (TestException ex) {

normal.assertSubscriptions(2);

}

}

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

public void mergeDelayErrorMultipleOneIsNull() {

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

c.await();

}

@Test(timeout = 5000)

public void mergeDelayErrorIterableEmpty() {