When designing any system, you also want to test it to guarantee its quality and that this quality does not regress as the system is modified throughout its lifetime. Ideally, you want to automate the process of testing it. Modern software is backed by thorough unit tests and Rx code should be no different.

Testing a synchronous piece of Rx code is as straight-forward as any unit test you are likely to find, using predefined sequences and [inspection](/Part 2 - Sequence Basics/3. Inspection.md). But what about asynchronous code? Consider testing the following piece of code:

Observable.interval(1, TimeUnit.SECONDS)

.take(5)

That is a sequence that takes 5 seconds to complete. That means every test that uses it will also take 5 seconds or more. That's not convenient at all if you have thousands of tests to run.

The piece of code above isn't just time consuming, it actually wastes all that time waiting for time to pass. If you could fast-forward the clock, that sequence would be evaluated almost instantly. Well, you can't fast-forward your system's clock, but Rx sees time through its schedulers. You can have a scheduler that virtualises time, called `TestScheduler`.

The `TestScheduler` does scheduling in the same way as the schedulers that we saw in the chapter about [Scheduling and threading](/Part 4 - Concurrency/1. Scheduling and threading.md). It schedules actions to be executed either immediately or in the future. The difference is that time is frozen and only progresses upon request.

As the name suggests, `advanceTimeTo` will execute all actions that are scheduled for up to a specific moment in time. That includes actions scheduled while the scheduler was being fast-forwarded, i.e. actions scheduled by other actions.

TestScheduler s = Schedulers.test();

s.createWorker().schedule(

() -> System.out.println("Immediate"));

s.createWorker().schedule(

() -> System.out.println("20s"),

20, TimeUnit.SECONDS);

s.createWorker().schedule(

() -> System.out.println("40s"),

40, TimeUnit.SECONDS);

System.out.println("Advancing to 1ms");

s.advanceTimeTo(1, TimeUnit.MILLISECONDS);

System.out.println("Virtual time: " + s.now());

System.out.println("Advancing to 10s");

s.advanceTimeTo(10, TimeUnit.SECONDS);

System.out.println("Virtual time: " + s.now());

System.out.println("Advancing to 40s");

s.advanceTimeTo(40, TimeUnit.SECONDS);

System.out.println("Virtual time: " + s.now());

Output

We scheduled 3 tasks: one to be executed immediately, and two to be executed in the future. Nothing happens until we advance time, including tasks scheduled immediately. When we advance time, tasks are executed from the queue, until a task that is not ready yet.

You can even set time to a previous moment than the one you are now. That isn't a useful feature as much as a source of bugs in your tests. Usually, `advanceTimeBy` will be closer to what you want to do.

`advanceTimeBy` advances time relative to the current moment in time. In every other regard, works like `advanceTimeTo`.

TestScheduler s = Schedulers.test();

s.createWorker().schedule(

() -> System.out.println("Immediate"));

s.createWorker().schedule(

() -> System.out.println("20s"),

20, TimeUnit.SECONDS);

s.createWorker().schedule(

() -> System.out.println("40s"),

40, TimeUnit.SECONDS);

System.out.println("Advancing by 1ms");

s.advanceTimeBy(1, TimeUnit.MILLISECONDS);

System.out.println("Virtual time: " + s.now());

System.out.println("Advancing by 10s");

s.advanceTimeBy(10, TimeUnit.SECONDS);

System.out.println("Virtual time: " + s.now());

System.out.println("Advancing by 40s");

s.advanceTimeBy(40, TimeUnit.SECONDS);

System.out.println("Virtual time: " + s.now());

Output

`triggerActions` does not advance time. It only executes actions that were scheduled to be executed up to the present.

TestScheduler s = Schedulers.test();

s.createWorker().schedule(

() -> System.out.println("Immediate"));

s.createWorker().schedule(

() -> System.out.println("20s"),

20, TimeUnit.SECONDS);

s.triggerActions();

System.out.println("Virtual time: " + s.now());

Output

There is nothing preventing actions from being scheduled for the same moment in time. When that happens, we have a scheduling collision. The order that two simultaneous tasks are executed is the same as the order in which they where scheduled.

TestScheduler s = Schedulers.test();

s.createWorker().schedule(

() -> System.out.println("First"),

20, TimeUnit.SECONDS);

s.createWorker().schedule(

() -> System.out.println("Second"),

20, TimeUnit.SECONDS);

s.createWorker().schedule(

() -> System.out.println("Third"),

20, TimeUnit.SECONDS);

s.advanceTimeTo(20, TimeUnit.SECONDS);

Output

Rx operators involving asynchronous, schedule those actions using a scheduler. If you take a look on all the operators in [Observable](http://reactivex.io/RxJava/javadoc/rx/Observable.html), you will see that such operators have overloads that take a scheduler. This is the way that you can supplement their real-time default schedulers for your `TestScheduler`.

Here is an example where we will test the output of `Observable.interval` against what we expect it to emit.

public void test() {

TestScheduler scheduler = new TestScheduler();

List<Long> expected = Arrays.asList(0L, 1L, 2L, 3L, 4L);

List<Long> result = new ArrayList<>();

Observable

.interval(1, TimeUnit.SECONDS, scheduler)

.take(5)

.subscribe(i -> result.add(i));

assertTrue(result.isEmpty());

scheduler.advanceTimeBy(5, TimeUnit.SECONDS);

assertTrue(result.equals(expected));

}

This is useful for testing small, self-contained pieces of Rx code, such as custom operators. A complete system may be using schedulers on its own, thus defeating our virtual time. [Lee Campbell suggests](http://www.introtorx.com/Content/v1.0.10621.0/16_TestingRx.html#TestingRx) abstracting over Rx's scheduler factories (`Schedulers`), with a provider of our own. When in debug-mode, our custom scheduler factory will replace all schedulers with a `TestScheduler`, which we will then use to control time throughout our system.

In the test above, we manually collected the values emitted and compared them against what we expected. This process is common enough in tests that Rx comes packaged with `TestScubscriber`, which will do that for us. It collectes every notification received. With `TestSubscriber` our previous test becomes:

public void test() {

TestScheduler scheduler = new TestScheduler();

TestSubscriber<Long> subscriber = new TestSubscriber<>();

List<Long> expected = Arrays.asList(0L, 1L, 2L, 3L, 4L);

Observable

.interval(1, TimeUnit.SECONDS, scheduler)

.take(5)

.subscribe(subscriber);

assertTrue(subscriber.getOnNextEvents().isEmpty());

scheduler.advanceTimeBy(5, TimeUnit.SECONDS);

subscriber.assertReceivedOnNext(expected);

}

A `TestSubscriber` collects more than just values and exposes them through the following methods:

java.lang.Thread getLastSeenThread()

java.util.List<Notification<T>> getOnCompletedEvents()

java.util.List<java.lang.Throwable> getOnErrorEvents()

java.util.List<T> getOnNextEvents()

There are two things to notice here. First is the `getLastSeenThread` method. A `TestSubscriber` checks on what thread it is notified and logs the most recent. That can be useful if, for example, you want to verify that an operation is/isn't executed on the GUI thread. Another interesting thing to notice is that there can be more than one termination event. That goes against how we defined our sequences in the begining of this guide. That is also the reason why the subscriber is capable of collecting multiple termination events: that would be a violation of the Rx contract and needs to be debugged.

`TestSubscriber` provides shorthands for a few basic assertions:

void assertNoErrors()

void assertReceivedOnNext(java.util.List<T> items)

void assertTerminalEvent()

void assertUnsubscribed()

There is also a way to block execution until the observable that the `TestSubscriber` is subscribed to terminates.

void awaitTerminalEvent()

void awaitTerminalEvent(long timeout, java.util.concurrent.TimeUnit unit)

void awaitTerminalEventAndUnsubscribeOnTimeout(long timeout, java.util.concurrent.TimeUnit unit)

Awaiting with a timeout will cause an exception if the observable fails to complete on time.