/**

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

import static org.junit.Assert.*;

import java.io.*;

import java.lang.Thread.UncaughtExceptionHandler;

import java.lang.reflect.*;

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.Scheduler.Worker;

import io.reactivex.disposables.*;

import io.reactivex.exceptions.*;

import io.reactivex.flowables.ConnectableFlowable;

import io.reactivex.functions.*;

import io.reactivex.internal.functions.Functions;

import io.reactivex.internal.operators.completable.CompletableError;

import io.reactivex.internal.operators.flowable.FlowableRange;

import io.reactivex.internal.operators.maybe.MaybeError;

import io.reactivex.internal.operators.observable.ObservableRange;

import io.reactivex.internal.operators.parallel.ParallelFromPublisher;

import io.reactivex.internal.operators.single.SingleJust;

import io.reactivex.internal.schedulers.ImmediateThinScheduler;

import io.reactivex.internal.subscriptions.ScalarSubscription;

import io.reactivex.observables.ConnectableObservable;

import io.reactivex.parallel.ParallelFlowable;

import io.reactivex.schedulers.Schedulers;

public class RxJavaPluginsTest {

// static Observable<Integer> createObservable() {

// return Observable.range(1, 5).map(new Function<Integer, Integer>() {

// @Override

// public Integer apply(Integer t) {

// throw new TestException();

// }

// });

// }

//

// static Flowable<Integer> createFlowable() {

// return Flowable.range(1, 5).map(new Function<Integer, Integer>() {

// @Override

// public Integer apply(Integer t) {

// throw new TestException();

// }

// });

// }

@Test

public void constructorShouldBePrivate() {

TestHelper.checkUtilityClass(RxJavaPlugins.class);

}

@Test

@Ignore("Not part of 2.0")

public void assemblyTrackingObservable() {

// RxJavaPlugins.enableAssemblyTracking();

// try {

// TestSubscriber<Integer> ts = TestSubscriber.create();

//

// createObservable().subscribe(ts);

//

// ts.assertError(TestException.class);

//

// Throwable ex = ts.getOnErrorEvents().get(0);

//

// AssemblyStackTraceException aste = AssemblyStackTraceException.find(ex);

//

// assertNotNull(aste);

//

// assertTrue(aste.getMessage(), aste.getMessage().contains("createObservable"));

//

// RxJavaPlugins.clearAssemblyTracking();

//

// ts = TestSubscriber.create();

//

// createObservable().subscribe(ts);

//

// ts.assertError(TestException.class);

//

// ex = ts.getOnErrorEvents().get(0);

//

// aste = AssemblyStackTraceException.find(ex);

//

// assertNull(aste);

// } finally {

// RxJavaPlugins.resetAssemblyTracking();

// }

}

// static Single<Integer> createSingle() {

// return Single.just(1).map(new Function<Integer, Integer>() {

// @Override

// public Integer apply(Integer t) {

// throw new TestException();

// }

// });

// }

@Test

@Ignore("Not part of 2.0")

public void assemblyTrackingSingle() {

// RxJavaPlugins.enableAssemblyTracking();

// try {

// TestSubscriber<Integer> ts = TestSubscriber.create();

//

// createSingle().subscribe(ts);

//

// ts.assertError(TestException.class);

//

// Throwable ex = ts.getOnErrorEvents().get(0);

//

// AssemblyStackTraceException aste = AssemblyStackTraceException.find(ex);

//

// assertNotNull(aste);

//

// assertTrue(aste.getMessage(), aste.getMessage().contains("createSingle"));

//

// RxJavaPlugins.clearAssemblyTracking();

//

// ts = TestSubscriber.create();

//

// createSingle().subscribe(ts);

//

// ts.assertError(TestException.class);

//

// ex = ts.getOnErrorEvents().get(0);

//

// aste = AssemblyStackTraceException.find(ex);

//

// assertNull(aste);

// } finally {

// RxJavaPlugins.resetAssemblyTracking();

// }

}

// static Completable createCompletable() {

// return Completable.error(new Callable<Throwable>() {

// @Override

// public Throwable call() {

// return new TestException();

// }

// });

// }

@Test

@Ignore("Not part of 2.0")

public void assemblyTrackingCompletable() {

// RxJavaPlugins.enableAssemblyTracking();

// try {

// TestSubscriber<Integer> ts = TestSubscriber.create();

//

// createCompletable().subscribe(ts);

//

// ts.assertError(TestException.class);

//

// Throwable ex = ts.getOnErrorEvents().get(0);

//

// AssemblyStackTraceException aste = AssemblyStackTraceException.find(ex);

//

// assertNotNull(aste);

//

// assertTrue(aste.getMessage(), aste.getMessage().contains("createCompletable"));

//

// RxJavaPlugins.clearAssemblyTracking();

//

// ts = TestSubscriber.create();

//

// createCompletable().subscribe(ts);

//

// ts.assertError(TestException.class);

//

// ex = ts.getOnErrorEvents().get(0);

//

// aste = AssemblyStackTraceException.find(ex);

//

// assertNull(aste);

//

// } finally {

// RxJavaPlugins.resetAssemblyTracking();

// }

}

@SuppressWarnings({ "rawtypes" })

@Test

public void lockdown() throws Exception {

RxJavaPlugins.reset();

RxJavaPlugins.lockdown();

try {

assertTrue(RxJavaPlugins.isLockdown());

Consumer a1 = Functions.emptyConsumer();

Callable f0 = new Callable() {

@Override

public Object call() {

return null;

}

};

Function f1 = Functions.identity();

BiFunction f2 = new BiFunction() {

@Override

public Object apply(Object t1, Object t2) {

return t2;

}

};

BooleanSupplier bs = new BooleanSupplier() {

@Override

public boolean getAsBoolean() throws Exception {

return true;

}

};

for (Method m : RxJavaPlugins.class.getMethods()) {

if (m.getName().startsWith("set")) {

Method getter;

Class<?> paramType = m.getParameterTypes()[0];

if (paramType == Boolean.TYPE) {

getter = RxJavaPlugins.class.getMethod("is" + m.getName().substring(3));

} else {

getter = RxJavaPlugins.class.getMethod("get" + m.getName().substring(3));

}

Object before = getter.invoke(null);

try {

if (paramType.isAssignableFrom(Boolean.TYPE)) {

m.invoke(null, true);

} else

if (paramType.isAssignableFrom(Callable.class)) {

m.invoke(null, f0);

} else

if (paramType.isAssignableFrom(Function.class)) {

m.invoke(null, f1);

} else

if (paramType.isAssignableFrom(Consumer.class)) {

m.invoke(null, a1);

} else

if (paramType.isAssignableFrom(BooleanSupplier.class)) {

m.invoke(null, bs);

} else {

m.invoke(null, f2);

}

fail("Should have thrown InvocationTargetException(IllegalStateException)");

} catch (InvocationTargetException ex) {

if (ex.getCause() instanceof IllegalStateException) {

assertEquals("Plugins can't be changed anymore", ex.getCause().getMessage());

} else {

fail("Should have thrown InvocationTargetException(IllegalStateException)");

}

}

Object after = getter.invoke(null);

if (paramType.isPrimitive()) {

assertEquals(m.toString(), before, after);

} else {

assertSame(m.toString(), before, after);

}

}

}

// Object o1 = RxJavaPlugins.getOnObservableCreate();

// Object o2 = RxJavaPlugins.getOnSingleCreate();

// Object o3 = RxJavaPlugins.getOnCompletableCreate();

//

// RxJavaPlugins.enableAssemblyTracking();

// RxJavaPlugins.clearAssemblyTracking();

// RxJavaPlugins.resetAssemblyTracking();

//

//

// assertSame(o1, RxJavaPlugins.getOnObservableCreate());

// assertSame(o2, RxJavaPlugins.getOnSingleCreate());

// assertSame(o3, RxJavaPlugins.getOnCompletableCreate());

} finally {

RxJavaPlugins.unlock();

RxJavaPlugins.reset();

assertFalse(RxJavaPlugins.isLockdown());

}

}

Function<Scheduler, Scheduler> replaceWithImmediate = new Function<Scheduler, Scheduler>() {

@Override

public Scheduler apply(Scheduler t) {

return ImmediateThinScheduler.INSTANCE;

}

};

@Test

public void overrideSingleScheduler() {

try {

RxJavaPlugins.setSingleSchedulerHandler(replaceWithImmediate);

assertSame(ImmediateThinScheduler.INSTANCE, Schedulers.single());

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

assertNotSame(ImmediateThinScheduler.INSTANCE, Schedulers.single());

}

@Test

public void overrideComputationScheduler() {

try {

RxJavaPlugins.setComputationSchedulerHandler(replaceWithImmediate);

assertSame(ImmediateThinScheduler.INSTANCE, Schedulers.computation());

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

assertNotSame(ImmediateThinScheduler.INSTANCE, Schedulers.computation());

}

@Test

public void overrideIoScheduler() {

try {

RxJavaPlugins.setIoSchedulerHandler(replaceWithImmediate);

assertSame(ImmediateThinScheduler.INSTANCE, Schedulers.io());

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

assertNotSame(ImmediateThinScheduler.INSTANCE, Schedulers.io());

}

@Test

public void overrideNewThreadScheduler() {

try {

RxJavaPlugins.setNewThreadSchedulerHandler(replaceWithImmediate);

assertSame(ImmediateThinScheduler.INSTANCE, Schedulers.newThread());

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

assertNotSame(ImmediateThinScheduler.INSTANCE, Schedulers.newThread());

}

Function<Callable<Scheduler>, Scheduler> initReplaceWithImmediate = new Function<Callable<Scheduler>, Scheduler>() {

@Override

public Scheduler apply(Callable<Scheduler> t) {

return ImmediateThinScheduler.INSTANCE;

}

};

@Test

public void overrideInitSingleScheduler() {

final Scheduler s = Schedulers.single(); // make sure the Schedulers is initialized

Callable<Scheduler> c = new Callable<Scheduler>() {

@Override

public Scheduler call() throws Exception {

return s;

}

};

try {

RxJavaPlugins.setInitSingleSchedulerHandler(initReplaceWithImmediate);

assertSame(ImmediateThinScheduler.INSTANCE, RxJavaPlugins.initSingleScheduler(c));

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

assertSame(s, RxJavaPlugins.initSingleScheduler(c));

}

@Test

public void overrideInitComputationScheduler() {

final Scheduler s = Schedulers.computation(); // make sure the Schedulers is initialized

Callable<Scheduler> c = new Callable<Scheduler>() {

@Override

public Scheduler call() throws Exception {

return s;

}

};

try {

RxJavaPlugins.setInitComputationSchedulerHandler(initReplaceWithImmediate);

assertSame(ImmediateThinScheduler.INSTANCE, RxJavaPlugins.initComputationScheduler(c));

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

assertSame(s, RxJavaPlugins.initComputationScheduler(c));

}

@Test

public void overrideInitIoScheduler() {

final Scheduler s = Schedulers.io(); // make sure the Schedulers is initialized;

Callable<Scheduler> c = new Callable<Scheduler>() {

@Override

public Scheduler call() throws Exception {

return s;

}

};

try {

RxJavaPlugins.setInitIoSchedulerHandler(initReplaceWithImmediate);

assertSame(ImmediateThinScheduler.INSTANCE, RxJavaPlugins.initIoScheduler(c));

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

assertSame(s, RxJavaPlugins.initIoScheduler(c));

}

@Test

public void overrideInitNewThreadScheduler() {

final Scheduler s = Schedulers.newThread(); // make sure the Schedulers is initialized;

Callable<Scheduler> c = new Callable<Scheduler>() {

@Override

public Scheduler call() throws Exception {

return s;

}

};

try {

RxJavaPlugins.setInitNewThreadSchedulerHandler(initReplaceWithImmediate);

assertSame(ImmediateThinScheduler.INSTANCE, RxJavaPlugins.initNewThreadScheduler(c));

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

assertSame(s, RxJavaPlugins.initNewThreadScheduler(c));

}

Callable<Scheduler> nullResultCallable = new Callable<Scheduler>() {

@Override

public Scheduler call() throws Exception {

return null;

}

};

@Test

public void overrideInitSingleSchedulerCrashes() {

// fail when Callable is null

try {

RxJavaPlugins.initSingleScheduler(null);

fail("Should have thrown NullPointerException");

} catch (NullPointerException npe) {

assertEquals("Scheduler Callable can't be null", npe.getMessage());

}

// fail when Callable result is null

try {

RxJavaPlugins.initSingleScheduler(nullResultCallable);

fail("Should have thrown NullPointerException");

} catch (NullPointerException npe) {

assertEquals("Scheduler Callable result can't be null", npe.getMessage());

}

}

@Test

public void overrideInitComputationSchedulerCrashes() {

// fail when Callable is null

try {

RxJavaPlugins.initComputationScheduler(null);

fail("Should have thrown NullPointerException");

} catch (NullPointerException npe) {

assertEquals("Scheduler Callable can't be null", npe.getMessage());

}

// fail when Callable result is null

try {

RxJavaPlugins.initComputationScheduler(nullResultCallable);

fail("Should have thrown NullPointerException");

} catch (NullPointerException npe) {

assertEquals("Scheduler Callable result can't be null", npe.getMessage());

}

}

@Test

public void overrideInitIoSchedulerCrashes() {

// fail when Callable is null

try {

RxJavaPlugins.initIoScheduler(null);

fail("Should have thrown NullPointerException");

} catch (NullPointerException npe) {

assertEquals("Scheduler Callable can't be null", npe.getMessage());

}

// fail when Callable result is null

try {

RxJavaPlugins.initIoScheduler(nullResultCallable);

fail("Should have thrown NullPointerException");

} catch (NullPointerException npe) {

assertEquals("Scheduler Callable result can't be null", npe.getMessage());

}

}

@Test

public void overrideInitNewThreadSchedulerCrashes() {

// fail when Callable is null

try {

RxJavaPlugins.initNewThreadScheduler(null);

fail("Should have thrown NullPointerException");

} catch (NullPointerException npe) {

// expected

assertEquals("Scheduler Callable can't be null", npe.getMessage());

}

// fail when Callable result is null

try {

RxJavaPlugins.initNewThreadScheduler(nullResultCallable);

fail("Should have thrown NullPointerException");

} catch (NullPointerException npe) {

assertEquals("Scheduler Callable result can't be null", npe.getMessage());

}

}

Callable<Scheduler> unsafeDefault = new Callable<Scheduler>() {

@Override

public Scheduler call() throws Exception {

throw new AssertionError("Default Scheduler instance should not have been evaluated");

}

};

@Test

public void testDefaultSingleSchedulerIsInitializedLazily() {

// unsafe default Scheduler Callable should not be evaluated

try {

RxJavaPlugins.setInitSingleSchedulerHandler(initReplaceWithImmediate);

RxJavaPlugins.initSingleScheduler(unsafeDefault);

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

assertNotSame(ImmediateThinScheduler.INSTANCE, Schedulers.single());

}

@Test

public void testDefaultIoSchedulerIsInitializedLazily() {

// unsafe default Scheduler Callable should not be evaluated

try {

RxJavaPlugins.setInitIoSchedulerHandler(initReplaceWithImmediate);

RxJavaPlugins.initIoScheduler(unsafeDefault);

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

assertNotSame(ImmediateThinScheduler.INSTANCE, Schedulers.io());

}

@Test

public void testDefaultComputationSchedulerIsInitializedLazily() {

// unsafe default Scheduler Callable should not be evaluated

try {

RxJavaPlugins.setInitComputationSchedulerHandler(initReplaceWithImmediate);

RxJavaPlugins.initComputationScheduler(unsafeDefault);

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

assertNotSame(ImmediateThinScheduler.INSTANCE, Schedulers.computation());

}

@Test

public void testDefaultNewThreadSchedulerIsInitializedLazily() {

// unsafe default Scheduler Callable should not be evaluated

try {

RxJavaPlugins.setInitNewThreadSchedulerHandler(initReplaceWithImmediate);

RxJavaPlugins.initNewThreadScheduler(unsafeDefault);

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

assertNotSame(ImmediateThinScheduler.INSTANCE, Schedulers.newThread());

}

@SuppressWarnings("rawtypes")

@Test

public void observableCreate() {

try {

RxJavaPlugins.setOnObservableAssembly(new Function<Observable, Observable>() {

@Override

public Observable apply(Observable t) {

return new ObservableRange(1, 2);

}

});

Observable.range(10, 3)

.test()

.assertValues(1, 2)

.assertNoErrors()

.assertComplete();

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

Observable.range(10, 3)

.test()

.assertValues(10, 11, 12)

.assertNoErrors()

.assertComplete();

}

@SuppressWarnings("rawtypes")

@Test

public void flowableCreate() {

try {

RxJavaPlugins.setOnFlowableAssembly(new Function<Flowable, Flowable>() {

@Override

public Flowable apply(Flowable t) {

return new FlowableRange(1, 2);

}

});

Flowable.range(10, 3)

.test()

.assertValues(1, 2)

.assertNoErrors()

.assertComplete();

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

Flowable.range(10, 3)

.test()

.assertValues(10, 11, 12)

.assertNoErrors()

.assertComplete();

}

@SuppressWarnings("rawtypes")

@Test

public void observableStart() {

try {

RxJavaPlugins.setOnObservableSubscribe(new BiFunction<Observable, Observer, Observer>() {

@Override

public Observer apply(Observable o, final Observer t) {

return new Observer() {

@Override

public void onSubscribe(Disposable d) {

t.onSubscribe(d);

}

@SuppressWarnings("unchecked")

@Override

public void onNext(Object value) {

t.onNext((Integer)value - 9);

}

@Override

public void onError(Throwable e) {

t.onError(e);

}

@Override

public void onComplete() {

t.onComplete();

}

};

}

});

Observable.range(10, 3)

.test()

.assertValues(1, 2, 3)

.assertNoErrors()

.assertComplete();

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

Observable.range(10, 3)

.test()

.assertValues(10, 11, 12)

.assertNoErrors()

.assertComplete();

}

@SuppressWarnings("rawtypes")

@Test

public void flowableStart() {

try {

RxJavaPlugins.setOnFlowableSubscribe(new BiFunction<Flowable, Subscriber, Subscriber>() {

@Override

public Subscriber apply(Flowable f, final Subscriber t) {

return new Subscriber() {

@Override

public void onSubscribe(Subscription s) {

t.onSubscribe(s);

}

@SuppressWarnings("unchecked")

@Override

public void onNext(Object value) {

t.onNext((Integer)value - 9);

}

@Override

public void onError(Throwable e) {

t.onError(e);

}

@Override

public void onComplete() {

t.onComplete();

}

};

}

});

Flowable.range(10, 3)

.test()

.assertValues(1, 2, 3)

.assertNoErrors()

.assertComplete();

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

Flowable.range(10, 3)

.test()

.assertValues(10, 11, 12)

.assertNoErrors()

.assertComplete();

}

@Test

@Ignore("Different architecture, no longer supported")

public void observableReturn() {

// try {

// final Subscription s = Subscriptions.empty();

//

// RxJavaPlugins.setOnObservableReturn(new Function<Subscription, Subscription>() {

// @Override

// public Subscription call(Subscription t) {

// return s;

// }

// });

//

// TestSubscriber<Integer> ts = TestSubscriber.create();

//

// Subscription u = Observable.range(10, 3).subscribe(ts);

//

// ts.assertValues(10, 11, 12);

// ts.assertNoErrors();

// ts.assertComplete();

//

// assertSame(s, u);

// } finally {

// RxJavaPlugins.reset();

// }

}

@SuppressWarnings("rawtypes")

@Test

public void singleCreate() {

try {

RxJavaPlugins.setOnSingleAssembly(new Function<Single, Single>() {

@Override

public Single apply(Single t) {

return new SingleJust<Integer>(10);

}

});

Single.just(1)

.test()

.assertValue(10)

.assertNoErrors()

.assertComplete();

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

Single.just(1)

.test()

.assertValue(1)

.assertNoErrors()

.assertComplete();

}

@SuppressWarnings("rawtypes")

@Test

public void singleStart() {

try {

RxJavaPlugins.setOnSingleSubscribe(new BiFunction<Single, SingleObserver, SingleObserver>() {

@Override

public SingleObserver apply(Single o, final SingleObserver t) {

return new SingleObserver<Object>() {

@Override

public void onSubscribe(Disposable d) {

t.onSubscribe(d);

}

@SuppressWarnings("unchecked")

@Override

public void onSuccess(Object value) {

t.onSuccess(10);

}

@Override

public void onError(Throwable e) {

t.onError(e);

}

};

}

});

Single.just(1)

.test()

.assertValue(10)

.assertNoErrors()

.assertComplete();

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

Single.just(1)

.test()

.assertValue(1)

.assertNoErrors()

.assertComplete();

}

@Test

@Ignore("Different architecture, no longer supported")

public void singleReturn() {

// try {

// final Subscription s = Subscriptions.empty();

//

// RxJavaPlugins.setOnSingleReturn(new Function<Subscription, Subscription>() {

// @Override

// public Subscription call(Subscription t) {

// return s;

// }

// });

//

// TestSubscriber<Integer> ts = TestSubscriber.create();

//

// Subscription u = Single.just(1).subscribe(ts);

//

// ts.assertValue(1);

// ts.assertNoErrors();

// ts.assertComplete();

//

// assertSame(s, u);

// } finally {

// RxJavaPlugins.reset();

// }

}

@Test

public void completableCreate() {

try {

RxJavaPlugins.setOnCompletableAssembly(new Function<Completable, Completable>() {

@Override

public Completable apply(Completable t) {

return new CompletableError(new TestException());

}

});

Completable.complete()

.test()

.assertNoValues()

.assertNotComplete()

.assertError(TestException.class);

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

Completable.complete()

.test()

.assertNoValues()

.assertNoErrors()

.assertComplete();

}

@Test

public void completableStart() {

try {

RxJavaPlugins.setOnCompletableSubscribe(new BiFunction<Completable, CompletableObserver, CompletableObserver>() {

@Override

public CompletableObserver apply(Completable o, final CompletableObserver t) {

return new CompletableObserver() {

@Override

public void onSubscribe(Disposable d) {

t.onSubscribe(d);

}

@Override

public void onError(Throwable e) {

t.onError(e);

}

@Override

public void onComplete() {

t.onError(new TestException());

}

};

}

});

Completable.complete()

.test()

.assertNoValues()

.assertNotComplete()

.assertError(TestException.class);

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

Completable.complete()

.test()

.assertNoValues()

.assertNoErrors()

.assertComplete();

}

void onSchedule(Worker w) throws InterruptedException {

try {

try {

final AtomicInteger value = new AtomicInteger();

final CountDownLatch cdl = new CountDownLatch(1);

RxJavaPlugins.setScheduleHandler(new Function<Runnable, Runnable>() {

@Override

public Runnable apply(Runnable t) {

return new Runnable() {

@Override

public void run() {

value.set(10);

cdl.countDown();

}

};

}

});

w.schedule(new Runnable() {

@Override

public void run() {

value.set(1);

cdl.countDown();

}

});

cdl.await();

assertEquals(10, value.get());

} finally {

RxJavaPlugins.reset();

}

// make sure the reset worked

final AtomicInteger value = new AtomicInteger();

final CountDownLatch cdl = new CountDownLatch(1);

w.schedule(new Runnable() {

@Override

public void run() {

value.set(1);

cdl.countDown();

}

});

cdl.await();

assertEquals(1, value.get());

} finally {