/**
* 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.parallel.ParallelFlowable;
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 the value type
* @return the mocked subscriber
*/
@SuppressWarnings("unchecked")
public static FlowableSubscriber mockSubscriber() {
FlowableSubscriber w = mock(FlowableSubscriber.class);
Mockito.doAnswer(new Answer() {
@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 the value type
* @return the mocked observer
*/
@SuppressWarnings("unchecked")
public static Observer mockObserver() {
return mock(Observer.class);
}
/**
* Mocks an MaybeObserver with the proper receiver type.
* @param the value type
* @return the mocked observer
*/
@SuppressWarnings("unchecked")
public static MaybeObserver mockMaybeObserver() {
return mock(MaybeObserver.class);
}
/**
* Mocks an SingleObserver with the proper receiver type.
* @param the value type
* @return the mocked observer
*/
@SuppressWarnings("unchecked")
public static SingleObserver 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 trackPluginErrors() {
final List list = Collections.synchronizedList(new ArrayList());
RxJavaPlugins.setErrorHandler(new Consumer() {
@Override
public void accept(Throwable t) {
list.add(t);
}
});
return list;
}
public static void assertError(List list, int index, Class ts, int index, Class ts, int index, Class ts, int index, Class ts, int index, Class source) {
List list = trackPluginErrors();
try {
final CountDownLatch cdl = new CountDownLatch(1);
source.subscribe(new FlowableSubscriber() {
@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.
* 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.
*
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 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 void checkEnum(Class enumClass) {
try {
Method m = enumClass.getMethod("values");
m.setAccessible(true);
Method e = enumClass.getMethod("valueOf", String.class);
m.setAccessible(true);
for (Enum o : (Enum[])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 the value type
* @param value the value not expected
* @return the consumer
*/
public static Consumer subscriberSingleNot(final T value) {
return new Consumer() {
@Override
public void accept(TestSubscriber 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 the value type
* @param value the value not expected
* @return the consumer
*/
public static Consumer observerSingleNot(final T value) {
return new Consumer() {
@Override
public void accept(TestObserver 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 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 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 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 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 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 the input value type
* @param source the source to test
*/
public static void checkDisposed(Flowable source) {
final TestSubscriber ts = new TestSubscriber(0L);
source.subscribe(new FlowableSubscriber() {
@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() {
@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() {
@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() {
@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, Observer, MaybeObserver, SingleObserver, 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 the input value type
* @param the output value type
* @param transform the transform to drive an operator
*/
public static void checkDoubleOnSubscribeMaybe(Function transform) {
List errors = trackPluginErrors();
try {
final Boolean[] b = { null, null };
final CountDownLatch cdl = new CountDownLatch(1);
Maybe source = new Maybe() {
@Override
protected void subscribeActual(MaybeObserver ts, Class ts, Object... classes) {
ts
.assertSubscribed()
.assertError(CompositeException.class)
.assertNotComplete();
List list = compositeList(ts.errors().get(0));
assertEquals(classes.length, list.size());
for (int i = 0; i < classes.length; i += 2) {
assertError(list, i, (Class)classes[i], (String)classes[i + 1]);
}
}
/**
* Check if the TestSubscriber has a CompositeException with the specified class
* of Throwables in the given order.
* @param ts the TestSubscriber instance
* @param classes the array of expected Throwables inside the Composite
*/
public static void assertCompositeExceptions(TestObserver ts, Class ts, Object... classes) {
ts
.assertSubscribed()
.assertError(CompositeException.class)
.assertNotComplete();
List list = compositeList(ts.errors().get(0));
assertEquals(classes.length, list.size());
for (int i = 0; i < classes.length; i += 2) {
assertError(list, i, (Class)classes[i], (String)classes[i + 1]);
}
}
/**
* Emit the given values and complete the Processor.
* @param the value type
* @param p the target processor
* @param values the values to emit
*/
public static void emit(Processor p, T... values) {
for (T v : values) {
p.onNext(v);
}
p.onComplete();
}
/**
* Emit the given values and complete the Subject.
* @param the value type
* @param p the target subject
* @param values the values to emit
*/
public static void emit(Subject p, T... values) {
for (T v : values) {
p.onNext(v);
}
p.onComplete();
}
/**
* Checks if the source is fuseable and its isEmpty/clear works properly.
* @param the value type
* @param source the source sequence
*/
public static void checkFusedIsEmptyClear(Observable source) {
final CountDownLatch cdl = new CountDownLatch(1);
final Boolean[] state = { null, null, null, null };
source.subscribe(new Observer() {
@Override
public void onSubscribe(Disposable d) {
try {
if (d instanceof QueueDisposable) {
@SuppressWarnings("unchecked")
QueueDisposable qd = (QueueDisposable) d;
state[0] = true;
int m = qd.requestFusion(QueueDisposable.ANY);
if (m != QueueDisposable.NONE) {
state[1] = true;
state[2] = qd.isEmpty();
qd.clear();
state[3] = qd.isEmpty();
}
}
cdl.countDown();
} finally {
d.dispose();
}
}
@Override
public void onNext(T value) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
try {
assertTrue(cdl.await(5, TimeUnit.SECONDS));
assertTrue("Not fuseable", state[0]);
assertTrue("Fusion rejected", state[1]);
assertNotNull(state[2]);
assertTrue("Did not empty", state[3]);
} catch (InterruptedException ex) {
throw new RuntimeException(ex);
}
}
/**
* Checks if the source is fuseable and its isEmpty/clear works properly.
* @param the value type
* @param source the source sequence
*/
public static void checkFusedIsEmptyClear(Flowable source) {
final CountDownLatch cdl = new CountDownLatch(1);
final Boolean[] state = { null, null, null, null };
source.subscribe(new FlowableSubscriber() {
@Override
public void onSubscribe(Subscription d) {
try {
if (d instanceof QueueSubscription) {
@SuppressWarnings("unchecked")
QueueSubscription qd = (QueueSubscription) d;
state[0] = true;
int m = qd.requestFusion(QueueSubscription.ANY);
if (m != QueueSubscription.NONE) {
state[1] = true;
state[2] = qd.isEmpty();
qd.clear();
state[3] = qd.isEmpty();
}
}
cdl.countDown();
} finally {
d.cancel();
}
}
@Override
public void onNext(T value) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
try {
assertTrue(cdl.await(5, TimeUnit.SECONDS));
assertTrue("Not fuseable", state[0]);
assertTrue("Fusion rejected", state[1]);
assertNotNull(state[2]);
assertTrue("Did not empty", state[3]);
} catch (InterruptedException ex) {
throw new RuntimeException(ex);
}
}
/**
* Returns an expanded error list of the given test consumer.
* @param to the test consumer instance
* @return the list
*/
public static List errorList(TestObserver to) {
return compositeList(to.errors().get(0));
}
/**
* Returns an expanded error list of the given test consumer.
* @param to the test consumer instance
* @return the list
*/
public static List errorList(TestSubscriber to) {
return compositeList(to.errors().get(0));
}
/**
* Tests the given mapping of a bad Observable by emitting the good values, then an error/completion and then
* a bad value followed by a TestException and and a completion.
* @param the value type
* @param mapper the mapper that receives a bad Observable and returns a reactive base type (detected via reflection).
* @param error if true, the good value emission is followed by a TestException("error"), if false then onComplete is called
* @param badValue the bad value to emit if not null
* @param goodValue the good value to emit before turning bad, if not null
* @param expected the expected resulting values, null to ignore values received
*/
public static void checkBadSourceObservable(Function mapper,
final boolean error, final T goodValue, final T badValue, final Object... expected) {
List errors = trackPluginErrors();
try {
Observable bad = new Observable() {
boolean once;
@Override
protected void subscribeActual(Observer) o;
TestObserver to = new TestObserver();
os.subscribe(to);
to.awaitDone(5, TimeUnit.SECONDS);
to.assertSubscribed();
if (expected != null) {
to.assertValues(expected);
}
if (error) {
to.assertError(TestException.class)
.assertErrorMessage("error")
.assertNotComplete();
} else {
to.assertNoErrors().assertComplete();
}
}
if (o instanceof Publisher) {
Publisher os = (Publisher) o;
TestSubscriber to = new TestSubscriber();
os.subscribe(to);
to.awaitDone(5, TimeUnit.SECONDS);
to.assertSubscribed();
if (expected != null) {
to.assertValues(expected);
}
if (error) {
to.assertError(TestException.class)
.assertErrorMessage("error")
.assertNotComplete();
} else {
to.assertNoErrors().assertComplete();
}
}
if (o instanceof SingleSource) {
SingleSource os = (SingleSource) o;
TestObserver to = new TestObserver();
os.subscribe(to);
to.awaitDone(5, TimeUnit.SECONDS);
to.assertSubscribed();
if (expected != null) {
to.assertValues(expected);
}
if (error) {
to.assertError(TestException.class)
.assertErrorMessage("error")
.assertNotComplete();
} else {
to.assertNoErrors().assertComplete();
}
}
if (o instanceof MaybeSource) {
MaybeSource os = (MaybeSource) o;
TestObserver to = new TestObserver();
os.subscribe(to);
to.awaitDone(5, TimeUnit.SECONDS);
to.assertSubscribed();
if (expected != null) {
to.assertValues(expected);
}
if (error) {
to.assertError(TestException.class)
.assertErrorMessage("error")
.assertNotComplete();
} else {
to.assertNoErrors().assertComplete();
}
}
if (o instanceof CompletableSource) {
CompletableSource os = (CompletableSource) o;
TestObserver to = new TestObserver();
os.subscribe(to);
to.awaitDone(5, TimeUnit.SECONDS);
to.assertSubscribed();
if (expected != null) {
to.assertValues(expected);
}
if (error) {
to.assertError(TestException.class)
.assertErrorMessage("error")
.assertNotComplete();
} else {
to.assertNoErrors().assertComplete();
}
}
assertUndeliverable(errors, 0, TestException.class, "second");
} catch (AssertionError ex) {
throw ex;
} catch (Throwable ex) {
throw new RuntimeException(ex);
} finally {
RxJavaPlugins.reset();
}
}
/**
* Tests the given mapping of a bad Observable by emitting the good values, then an error/completion and then
* a bad value followed by a TestException and and a completion.
* @param the value type
* @param mapper the mapper that receives a bad Observable and returns a reactive base type (detected via reflection).
* @param error if true, the good value emission is followed by a TestException("error"), if false then onComplete is called
* @param badValue the bad value to emit if not null
* @param goodValue the good value to emit before turning bad, if not null
* @param expected the expected resulting values, null to ignore values received
*/
public static void checkBadSourceFlowable(Function mapper,
final boolean error, final T goodValue, final T badValue, final Object... expected) {
List errors = trackPluginErrors();
try {
Flowable bad = new Flowable() {
@Override
protected void subscribeActual(Subscriber) o;
TestObserver to = new TestObserver();
os.subscribe(to);
to.awaitDone(5, TimeUnit.SECONDS);
to.assertSubscribed();
if (expected != null) {
to.assertValues(expected);
}
if (error) {
to.assertError(TestException.class)
.assertErrorMessage("error")
.assertNotComplete();
} else {
to.assertNoErrors().assertComplete();
}
}
if (o instanceof Publisher) {
Publisher os = (Publisher) o;
TestSubscriber to = new TestSubscriber();
os.subscribe(to);
to.awaitDone(5, TimeUnit.SECONDS);
to.assertSubscribed();
if (expected != null) {
to.assertValues(expected);
}
if (error) {
to.assertError(TestException.class)
.assertErrorMessage("error")
.assertNotComplete();
} else {
to.assertNoErrors().assertComplete();
}
}
if (o instanceof SingleSource) {
SingleSource os = (SingleSource) o;
TestObserver to = new TestObserver();
os.subscribe(to);
to.awaitDone(5, TimeUnit.SECONDS);
to.assertSubscribed();
if (expected != null) {
to.assertValues(expected);
}
if (error) {
to.assertError(TestException.class)
.assertErrorMessage("error")
.assertNotComplete();
} else {
to.assertNoErrors().assertComplete();
}
}
if (o instanceof MaybeSource) {
MaybeSource os = (MaybeSource) o;
TestObserver to = new TestObserver();
os.subscribe(to);
to.awaitDone(5, TimeUnit.SECONDS);
to.assertSubscribed();
if (expected != null) {
to.assertValues(expected);
}
if (error) {
to.assertError(TestException.class)
.assertErrorMessage("error")
.assertNotComplete();
} else {
to.assertNoErrors().assertComplete();
}
}
if (o instanceof CompletableSource) {
CompletableSource os = (CompletableSource) o;
TestObserver to = new TestObserver();
os.subscribe(to);
to.awaitDone(5, TimeUnit.SECONDS);
to.assertSubscribed();
if (expected != null) {
to.assertValues(expected);
}
if (error) {
to.assertError(TestException.class)
.assertErrorMessage("error")
.assertNotComplete();
} else {
to.assertNoErrors().assertComplete();
}
}
assertUndeliverable(errors, 0, TestException.class, "second");
} catch (AssertionError ex) {
throw ex;
} catch (Throwable ex) {
throw new RuntimeException(ex);
} finally {
RxJavaPlugins.reset();
}
}
public static void checkInvalidParallelSubscribers(ParallelFlowable source) {
int n = source.parallelism();
@SuppressWarnings("unchecked")
TestSubscriber[] tss = new TestSubscriber[n + 1];
for (int i = 0; i