RxJava test Observable

Question

There is something like TestSubscriber. It has very useful functions like awaitTerminalEvent(), getOnNextEvents(), etc.

Is there an equivalent of TestSubscriber for Observable that might be pinned to some observable in the middle of sequence?

I would like to test some specific cases, one of which is:

• I have a class that accepts Observable in constructor. One of it's methods invokes that Observable inside onErrorResumeNext operator. So executing the main Observable under specific conditions might invoke tested Observable. Something like assertExecuted is needed.

• In the case described above I would like to test Thread the Observable is executed on.

• I would like to know, what Observable returned based on it's input.

Each of those cases require workarounds like:

observable
  .doOnNext(o -> {
      // Observable was given a value
   });

If there is no nice way already available, how to make the call above cleaner?

Answer 1

Assertions need to be tested outside of the lambdas that are part of the observable because the lambdas themselves may never run (like in your onErrorResumeNext case). So you need some object defined outside of the observable to carry the result that we want to test.

A common technique for testing observables is to use Atomic objects to record events:

AtomicBoolean executed = new AtomicBoolean(false);
Observable
  .empty()
  .doOnNext(t -> executed.set(true))
  .subscribe();
assertFalse(executed.get());

This example could be better done with TestSubscriber but it shows the technique.

rxjava-extras has convenience classes if you are using java 6 (heck just use retrolambda) so that you don't need to write anonymous classes for this use case:

AtomicBoolean executed = new AtomicBoolean(false);
Observable
  .empty()
  .doOnNext(Actions.setToTrue(executed))
  .subscribe();
assertFalse(executed.get());