/**

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

import static org.junit.Assert.*;

import static org.mockito.ArgumentMatchers.any;

import static org.mockito.Mockito.mock;

import java.lang.reflect.*;

import java.util.*;

import java.util.concurrent.*;

import java.util.concurrent.atomic.AtomicInteger;

import org.mockito.Mockito;

import org.mockito.invocation.InvocationOnMock;

import org.mockito.stubbing.Answer;

import org.reactivestreams.*;

import io.reactivex.disposables.*;

import io.reactivex.exceptions.*;

import io.reactivex.functions.*;

import io.reactivex.internal.functions.ObjectHelper;

import io.reactivex.internal.fuseable.*;

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

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

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

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.Schedulers;

import io.reactivex.subjects.Subject;

import io.reactivex.subscribers.TestSubscriber;

/**

* Common methods for helping with tests from 1.x mostly.

*/

public enum TestHelper {

;

/**

* Mocks a subscriber and prepares it to request Long.MAX_VALUE.

* @param <T> the value type

* @return the mocked subscriber

*/

@SuppressWarnings("unchecked")

public static <T> FlowableSubscriber<T> mockSubscriber() {

FlowableSubscriber<T> w = mock(FlowableSubscriber.class);

Mockito.doAnswer(new Answer<Object>() {

@Override

public Object answer(InvocationOnMock a) throws Throwable {

Subscription s = a.getArgument(0);

s.request(Long.MAX_VALUE);

return null;

}

}).when(w).onSubscribe((Subscription)any());

return w;

}

/**

* Mocks an Observer with the proper receiver type.

* @param <T> the value type

* @return the mocked observer

*/

@SuppressWarnings("unchecked")

public static <T> Observer<T> mockObserver() {

return mock(Observer.class);

}

/**

* Mocks an MaybeObserver with the proper receiver type.

* @param <T> the value type

* @return the mocked observer

*/

@SuppressWarnings("unchecked")

public static <T> MaybeObserver<T> mockMaybeObserver() {

return mock(MaybeObserver.class);

}

/**

* Mocks an SingleObserver with the proper receiver type.

* @param <T> the value type

* @return the mocked observer

*/

@SuppressWarnings("unchecked")

public static <T> SingleObserver<T> mockSingleObserver() {

return mock(SingleObserver.class);

}

/**

* Mocks an CompletableObserver.

* @return the mocked observer

*/

public static CompletableObserver mockCompletableObserver() {

return mock(CompletableObserver.class);

}

/**

* Validates that the given class, when forcefully instantiated throws

* an IllegalArgumentException("No instances!") exception.

* @param clazz the class to test, not null

*/

public static void checkUtilityClass(Class<?> clazz) {

try {

Constructor<?> c = clazz.getDeclaredConstructor();

c.setAccessible(true);

try {

c.newInstance();

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

} catch (InvocationTargetException ex) {

assertEquals("No instances!", ex.getCause().getMessage());

}

} catch (Exception ex) {

AssertionError ae = new AssertionError(ex.toString());

ae.initCause(ex);

throw ae;

}

}

public static List<Throwable> trackPluginErrors() {

final List<Throwable> list = Collections.synchronizedList(new ArrayList<Throwable>());

RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {

@Override

public void accept(Throwable t) {

list.add(t);

}

});

return list;

}

public static void assertError(List<Throwable> list, int index, Class<? extends Throwable> clazz) {

Throwable ex = list.get(index);

if (!clazz.isInstance(ex)) {

AssertionError err = new AssertionError(clazz + " expected but got " + list.get(index));

err.initCause(list.get(index));

throw err;

}

}

public static void assertUndeliverable(List<Throwable> list, int index, Class<? extends Throwable> clazz) {

Throwable ex = list.get(index);

if (!(ex instanceof UndeliverableException)) {

AssertionError err = new AssertionError("Outer exception UndeliverableException expected but got " + list.get(index));

err.initCause(list.get(index));

throw err;

}

ex = ex.getCause();

if (!clazz.isInstance(ex)) {

AssertionError err = new AssertionError("Inner exception " + clazz + " expected but got " + list.get(index));

err.initCause(list.get(index));

throw err;

}

}

public static void assertError(List<Throwable> list, int index, Class<? extends Throwable> clazz, String message) {

Throwable ex = list.get(index);

if (!clazz.isInstance(ex)) {

AssertionError err = new AssertionError("Type " + clazz + " expected but got " + ex);

err.initCause(ex);

throw err;

}

if (!ObjectHelper.equals(message, ex.getMessage())) {

AssertionError err = new AssertionError("Message " + message + " expected but got " + ex.getMessage());

err.initCause(ex);

throw err;

}

}

public static void assertUndeliverable(List<Throwable> list, int index, Class<? extends Throwable> clazz, String message) {

Throwable ex = list.get(index);

if (!(ex instanceof UndeliverableException)) {

AssertionError err = new AssertionError("Outer exception UndeliverableException expected but got " + list.get(index));

err.initCause(list.get(index));

throw err;

}

ex = ex.getCause();

if (!clazz.isInstance(ex)) {

AssertionError err = new AssertionError("Inner exception " + clazz + " expected but got " + list.get(index));

err.initCause(list.get(index));

throw err;

}

if (!ObjectHelper.equals(message, ex.getMessage())) {

AssertionError err = new AssertionError("Message " + message + " expected but got " + ex.getMessage());

err.initCause(ex);

throw err;

}

}

public static void assertError(TestObserver<?> ts, int index, Class<? extends Throwable> clazz) {

Throwable ex = ts.errors().get(0);

try {

if (ex instanceof CompositeException) {

CompositeException ce = (CompositeException) ex;

List<Throwable> cel = ce.getExceptions();

assertTrue(cel.get(index).toString(), clazz.isInstance(cel.get(index)));

} else {

fail(ex.toString() + ": not a CompositeException");

}

} catch (AssertionError e) {

ex.printStackTrace();

throw e;

}

}

public static void assertError(TestSubscriber<?> ts, int index, Class<? extends Throwable> clazz) {

Throwable ex = ts.errors().get(0);

if (ex instanceof CompositeException) {

CompositeException ce = (CompositeException) ex;

List<Throwable> cel = ce.getExceptions();

assertTrue(cel.get(index).toString(), clazz.isInstance(cel.get(index)));

} else {

fail(ex.toString() + ": not a CompositeException");

}

}

public static void assertError(TestObserver<?> ts, int index, Class<? extends Throwable> clazz, String message) {

Throwable ex = ts.errors().get(0);

if (ex instanceof CompositeException) {

CompositeException ce = (CompositeException) ex;

List<Throwable> cel = ce.getExceptions();

assertTrue(cel.get(index).toString(), clazz.isInstance(cel.get(index)));

assertEquals(message, cel.get(index).getMessage());

} else {

fail(ex.toString() + ": not a CompositeException");

}

}

public static void assertError(TestSubscriber<?> ts, int index, Class<? extends Throwable> clazz, String message) {

Throwable ex = ts.errors().get(0);

if (ex instanceof CompositeException) {

CompositeException ce = (CompositeException) ex;

List<Throwable> cel = ce.getExceptions();

assertTrue(cel.get(index).toString(), clazz.isInstance(cel.get(index)));

assertEquals(message, cel.get(index).getMessage());

} else {

fail(ex.toString() + ": not a CompositeException");

}

}

/**

* Verify that a specific enum type has no enum constants.

* @param <E> the enum type

* @param e the enum class instance

*/

public static <E extends Enum<E>> void assertEmptyEnum(Class<E> e) {

assertEquals(0, e.getEnumConstants().length);

try {

try {

Method m0 = e.getDeclaredMethod("values");

Object[] a = (Object[])m0.invoke(null);

assertEquals(0, a.length);

Method m = e.getDeclaredMethod("valueOf", String.class);

m.invoke("INSTANCE");

fail("Should have thrown!");

} catch (InvocationTargetException ex) {

fail(ex.toString());

} catch (IllegalAccessException ex) {

fail(ex.toString());

} catch (IllegalArgumentException ex) {

// we expected this

}

} catch (NoSuchMethodException ex) {

fail(ex.toString());

}

}

/**

* Assert that by consuming the Publisher with a bad request amount, it is

* reported to the plugin error handler promptly.

* @param source the source to consume

*/

public static void assertBadRequestReported(Publisher<?> source) {

List<Throwable> list = trackPluginErrors();

try {

final CountDownLatch cdl = new CountDownLatch(1);

source.subscribe(new FlowableSubscriber<Object>() {

@Override

public void onSubscribe(Subscription s) {

try {

s.request(-99);

s.cancel();

s.cancel();

} finally {

cdl.countDown();

}

}

@Override

public void onNext(Object t) {

}

@Override

public void onError(Throwable t) {

}

@Override

public void onComplete() {

}

});

try {

assertTrue(cdl.await(5, TimeUnit.SECONDS));

} catch (InterruptedException ex) {

throw new AssertionError(ex.getMessage());

}

assertTrue(list.toString(), list.get(0) instanceof IllegalArgumentException);

assertEquals("n > 0 required but it was -99", list.get(0).getMessage());

} finally {

RxJavaPlugins.setErrorHandler(null);

}

}

/**

* Synchronizes the execution of two runnables (as much as possible)

* to test race conditions.

* <p>The method blocks until both have run to completion.

* @param r1 the first runnable

* @param r2 the second runnable

*/

public static void race(final Runnable r1, final Runnable r2) {

race(r1, r2, Schedulers.single());

}

/**

* Synchronizes the execution of two runnables (as much as possible)

* to test race conditions.

* <p>The method blocks until both have run to completion.

* @param r1 the first runnable

* @param r2 the second runnable

* @param s the scheduler to use

*/

public static void race(final Runnable r1, final Runnable r2, Scheduler s) {

final AtomicInteger count = new AtomicInteger(2);

final CountDownLatch cdl = new CountDownLatch(2);

final Throwable[] errors = { null, null };

s.scheduleDirect(new Runnable() {

@Override

public void run() {

if (count.decrementAndGet() != 0) {

while (count.get() != 0) { }

}

try {

try {

r1.run();

} catch (Throwable ex) {

errors[0] = ex;

}

} finally {

cdl.countDown();

}

}

});

if (count.decrementAndGet() != 0) {

while (count.get() != 0) { }

}

try {

try {

r2.run();

} catch (Throwable ex) {

errors[1] = ex;

}

} finally {

cdl.countDown();

}

try {

if (!cdl.await(5, TimeUnit.SECONDS)) {

throw new AssertionError("The wait timed out!");

}

} catch (InterruptedException ex) {

throw new RuntimeException(ex);

}

if (errors[0] != null && errors[1] == null) {

throw ExceptionHelper.wrapOrThrow(errors[0]);

}

if (errors[0] == null && errors[1] != null) {

throw ExceptionHelper.wrapOrThrow(errors[1]);

}

if (errors[0] != null && errors[1] != null) {

throw new CompositeException(errors);

}

}

/**

* Cast the given Throwable to CompositeException and returns its inner

* Throwable list.

* @param ex the target Throwable

* @return the list of Throwables

*/

public static List<Throwable> compositeList(Throwable ex) {

if (ex instanceof UndeliverableException) {

ex = ex.getCause();

}

return ((CompositeException)ex).getExceptions();

}

/**

* Assert that the offer methods throw UnsupportedOperationExcetpion.

* @param q the queue implementation

*/

public static void assertNoOffer(SimpleQueue<?> q) {

try {

q.offer(null);

fail("Should have thrown!");

} catch (UnsupportedOperationException ex) {

// expected

}

try {

q.offer(null, null);

fail("Should have thrown!");

} catch (UnsupportedOperationException ex) {

// expected

}

}

@SuppressWarnings("unchecked")

public static <E extends Enum<E>> void checkEnum(Class<E> enumClass) {

try {

Method m = enumClass.getMethod("values");

m.setAccessible(true);

Method e = enumClass.getMethod("valueOf", String.class);

m.setAccessible(true);

for (Enum<E> o : (Enum<E>[])m.invoke(null)) {

assertSame(o, e.invoke(null, o.name()));

}

} catch (Throwable ex) {

throw ExceptionHelper.wrapOrThrow(ex);

}

}

/**

* Returns an Consumer that asserts the TestSubscriber has exaclty one value + completed

* normally and that single value is not the value specified.

* @param <T> the value type

* @param value the value not expected

* @return the consumer

*/

public static <T> Consumer<TestSubscriber<T>> subscriberSingleNot(final T value) {

return new Consumer<TestSubscriber<T>>() {

@Override

public void accept(TestSubscriber<T> ts) throws Exception {

ts

.assertSubscribed()

.assertValueCount(1)

.assertNoErrors()

.assertComplete();

T v = ts.values().get(0);

assertNotEquals(value, v);

}

};

}

/**

* Returns an Consumer that asserts the TestObserver has exaclty one value + completed

* normally and that single value is not the value specified.

* @param <T> the value type

* @param value the value not expected

* @return the consumer

*/

public static <T> Consumer<TestObserver<T>> observerSingleNot(final T value) {

return new Consumer<TestObserver<T>>() {

@Override

public void accept(TestObserver<T> ts) throws Exception {

ts

.assertSubscribed()

.assertValueCount(1)

.assertNoErrors()

.assertComplete();

T v = ts.values().get(0);

assertNotEquals(value, v);

}

};

}

/**

* Calls onSubscribe twice and checks if it doesn't affect the first Subscription while

* reporting it to plugin error handler.

* @param subscriber the target

*/

public static void doubleOnSubscribe(Subscriber<?> subscriber) {

List<Throwable> errors = trackPluginErrors();

try {

BooleanSubscription s1 = new BooleanSubscription();

subscriber.onSubscribe(s1);

BooleanSubscription s2 = new BooleanSubscription();

subscriber.onSubscribe(s2);

assertFalse(s1.isCancelled());

assertTrue(s2.isCancelled());

assertError(errors, 0, IllegalStateException.class, "Subscription already set!");

} finally {

RxJavaPlugins.reset();

}

}

/**

* Calls onSubscribe twice and checks if it doesn't affect the first Disposable while

* reporting it to plugin error handler.

* @param subscriber the target

*/

public static void doubleOnSubscribe(Observer<?> subscriber) {

List<Throwable> errors = trackPluginErrors();

try {

Disposable d1 = Disposables.empty();

subscriber.onSubscribe(d1);

Disposable d2 = Disposables.empty();

subscriber.onSubscribe(d2);

assertFalse(d1.isDisposed());

assertTrue(d2.isDisposed());

assertError(errors, 0, IllegalStateException.class, "Disposable already set!");

} finally {

RxJavaPlugins.reset();

}

}

/**

* Calls onSubscribe twice and checks if it doesn't affect the first Disposable while

* reporting it to plugin error handler.

* @param subscriber the target

*/

public static void doubleOnSubscribe(SingleObserver<?> subscriber) {

List<Throwable> errors = trackPluginErrors();

try {

Disposable d1 = Disposables.empty();

subscriber.onSubscribe(d1);

Disposable d2 = Disposables.empty();

subscriber.onSubscribe(d2);

assertFalse(d1.isDisposed());

assertTrue(d2.isDisposed());

assertError(errors, 0, IllegalStateException.class, "Disposable already set!");

} finally {

RxJavaPlugins.reset();

}

}

/**

* Calls onSubscribe twice and checks if it doesn't affect the first Disposable while

* reporting it to plugin error handler.

* @param subscriber the target

*/

public static void doubleOnSubscribe(CompletableObserver subscriber) {

List<Throwable> errors = trackPluginErrors();

try {

Disposable d1 = Disposables.empty();

subscriber.onSubscribe(d1);

Disposable d2 = Disposables.empty();

subscriber.onSubscribe(d2);

assertFalse(d1.isDisposed());

assertTrue(d2.isDisposed());

assertError(errors, 0, IllegalStateException.class, "Disposable already set!");

} finally {

RxJavaPlugins.reset();

}

}

/**

* Calls onSubscribe twice and checks if it doesn't affect the first Disposable while

* reporting it to plugin error handler.

* @param subscriber the target

*/

public static void doubleOnSubscribe(MaybeObserver<?> subscriber) {

List<Throwable> errors = trackPluginErrors();

try {

Disposable d1 = Disposables.empty();

subscriber.onSubscribe(d1);

Disposable d2 = Disposables.empty();

subscriber.onSubscribe(d2);

assertFalse(d1.isDisposed());

assertTrue(d2.isDisposed());

assertError(errors, 0, IllegalStateException.class, "Disposable already set!");

} finally {

RxJavaPlugins.reset();

}

}

/**

* Checks if the upstream's Subscription sent through the onSubscribe reports

* isCancelled properly before and after calling dispose.

* @param <T> the input value type

* @param source the source to test

*/

public static <T> void checkDisposed(Flowable<T> source) {

final TestSubscriber<Object> ts = new TestSubscriber<Object>(0L);

source.subscribe(new FlowableSubscriber<Object>() {

@Override

public void onSubscribe(Subscription s) {

ts.onSubscribe(new BooleanSubscription());

s.cancel();

s.cancel();

}

@Override

public void onNext(Object t) {

ts.onNext(t);

}

@Override

public void onError(Throwable t) {

ts.onError(t);

}

@Override

public void onComplete() {

ts.onComplete();

}

});

ts.assertEmpty();

}

/**

* Checks if the upstream's Disposable sent through the onSubscribe reports

* isDisposed properly before and after calling dispose.

* @param source the source to test

*/

public static void checkDisposed(Maybe<?> source) {

final Boolean[] b = { null, null };

final CountDownLatch cdl = new CountDownLatch(1);

source.subscribe(new MaybeObserver<Object>() {

@Override

public void onSubscribe(Disposable d) {

try {

b[0] = d.isDisposed();

d.dispose();

b[1] = d.isDisposed();

d.dispose();

} finally {

cdl.countDown();

}

}

@Override

public void onSuccess(Object value) {

// ignored

}

@Override

public void onError(Throwable e) {

// ignored

}

@Override

public void onComplete() {

// ignored

}

});

try {

assertTrue("Timed out", cdl.await(5, TimeUnit.SECONDS));

} catch (InterruptedException ex) {

throw ExceptionHelper.wrapOrThrow(ex);

}

assertEquals("Reports disposed upfront?", false, b[0]);

assertEquals("Didn't report disposed after?", true, b[1]);

}

/**

* Checks if the upstream's Disposable sent through the onSubscribe reports

* isDisposed properly before and after calling dispose.

* @param source the source to test

*/

public static void checkDisposed(Observable<?> source) {

final Boolean[] b = { null, null };

final CountDownLatch cdl = new CountDownLatch(1);

source.subscribe(new Observer<Object>() {

@Override

public void onSubscribe(Disposable d) {

try {

b[0] = d.isDisposed();

d.dispose();

b[1] = d.isDisposed();

d.dispose();

} finally {

cdl.countDown();

}

}

@Override

public void onNext(Object value) {

// ignored

}

@Override

public void onError(Throwable e) {

// ignored

}

@Override

public void onComplete() {

// ignored

}

});

try {

assertTrue("Timed out", cdl.await(5, TimeUnit.SECONDS));

} catch (InterruptedException ex) {

throw ExceptionHelper.wrapOrThrow(ex);

}

assertEquals("Reports disposed upfront?", false, b[0]);

assertEquals("Didn't report disposed after?", true, b[1]);

}

/**

* Checks if the upstream's Disposable sent through the onSubscribe reports

* isDisposed properly before and after calling dispose.

* @param source the source to test

*/

public static void checkDisposed(Single<?> source) {

final Boolean[] b = { null, null };

final CountDownLatch cdl = new CountDownLatch(1);

source.subscribe(new SingleObserver<Object>() {

@Override

public void onSubscribe(Disposable d) {

try {

b[0] = d.isDisposed();

d.dispose();

b[1] = d.isDisposed();

d.dispose();

} finally {

cdl.countDown();

}

}

@Override

public void onSuccess(Object value) {

// ignored

}

@Override

public void onError(Throwable e) {

// ignored

}

});

try {

assertTrue("Timed out", cdl.await(5, TimeUnit.SECONDS));

} catch (InterruptedException ex) {

throw ExceptionHelper.wrapOrThrow(ex);

}

assertEquals("Reports disposed upfront?", false, b[0]);

assertEquals("Didn't report disposed after?", true, b[1]);

}

/**

* Checks if the upstream's Disposable sent through the onSubscribe reports

* isDisposed properly before and after calling dispose.

* @param source the source to test

*/

public static void checkDisposed(Completable source) {

final Boolean[] b = { null, null };

final CountDownLatch cdl = new CountDownLatch(1);

source.subscribe(new CompletableObserver() {

@Override

public void onSubscribe(Disposable d) {

try {

b[0] = d.isDisposed();

d.dispose();

b[1] = d.isDisposed();

d.dispose();

} finally {

cdl.countDown();

}

}

@Override

public void onError(Throwable e) {

// ignored

}

@Override

public void onComplete() {

// ignored

}

});

try {

assertTrue("Timed out", cdl.await(5, TimeUnit.SECONDS));

} catch (InterruptedException ex) {

throw ExceptionHelper.wrapOrThrow(ex);

}

assertEquals("Reports disposed upfront?", false, b[0]);

assertEquals("Didn't report disposed after?", true, b[1]);

}

/**

* Consumer for all base reactive types.

*/

enum NoOpConsumer implements FlowableSubscriber<Object>, Observer<Object>, MaybeObserver<Object>, SingleObserver<Object>, CompletableObserver {

INSTANCE;

@Override

public void onSubscribe(Disposable d) {

// deliberately no-op

}

@Override

public void onSuccess(Object value) {

// deliberately no-op

}

@Override

public void onError(Throwable e) {

// deliberately no-op

}

@Override

public void onComplete() {

// deliberately no-op

}

@Override

public void onSubscribe(Subscription s) {

// deliberately no-op

}

@Override

public void onNext(Object t) {

// deliberately no-op

}

}

/**

* Check if the given transformed reactive type reports multiple onSubscribe calls to

* RxJavaPlugins.

* @param <T> the input value type

* @param <R> the output value type

* @param transform the transform to drive an operator

*/

public static <T, R> void checkDoubleOnSubscribeMaybe(Function<Maybe<T>, ? extends MaybeSource<R>> transform) {

List<Throwable> errors = trackPluginErrors();

try {

final Boolean[] b = { null, null };

final CountDownLatch cdl = new CountDownLatch(1);

Maybe<T> source = new Maybe<T>() {

@Override

protected void subscribeActual(MaybeObserver<? super T> observer) {

try {

Disposable d1 = Disposables.empty();

observer.onSubscribe(d1);

Disposable d2 = Disposables.empty();

observer.onSubscribe(d2);

b[0] = d1.isDisposed();

b[1] = d2.isDisposed();

} finally {

cdl.countDown();

}

}

};

MaybeSource<R> out = transform.apply(source);

out.subscribe(NoOpConsumer.INSTANCE);

try {

assertTrue("Timed out", cdl.await(5, TimeUnit.SECONDS));

} catch (InterruptedException ex) {

throw ExceptionHelper.wrapOrThrow(ex);

}

assertEquals("First disposed?", false, b[0]);

assertEquals("Second not disposed?", true, b[1]);

assertError(errors, 0, IllegalStateException.class, "Disposable already set!");

} catch (Throwable ex) {

throw ExceptionHelper.wrapOrThrow(ex);

} finally {

RxJavaPlugins.reset();

}

}

/**

* Check if the given transformed reactive type reports multiple onSubscribe calls to

* RxJavaPlugins.

* @param <T> the input value type

* @param <R> the output value type

* @param transform the transform to drive an operator

*/

public static <T, R> void checkDoubleOnSubscribeMaybeToSingle(Function<Maybe<T>, ? extends SingleSource<R>> transform) {

List<Throwable> errors = trackPluginErrors();

try {

final Boolean[] b = { null, null };

final CountDownLatch cdl = new CountDownLatch(1);

Maybe<T> source = new Maybe<T>() {

@Override

protected void subscribeActual(MaybeObserver<? super T> observer) {

try {

Disposable d1 = Disposables.empty();

observer.onSubscribe(d1);

Disposable d2 = Disposables.empty();

observer.onSubscribe(d2);

b[0] = d1.isDisposed();

b[1] = d2.isDisposed();

} finally {

cdl.countDown();

}

}

};

SingleSource<R> out = transform.apply(source);

out.subscribe(NoOpConsumer.INSTANCE);

try {

assertTrue("Timed out", cdl.await(5, TimeUnit.SECONDS));