When designing any system, you want guarantees about the correctness of the operations and that this quality does not regress as the system is modified throughout its lifetime. You'll design tests, which, ideally, should be automated. 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 this sequence will 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 of that time doing nothing while waiting. If you could fast-forward the clock, that sequence would be evaluated almost instantly. You can't actually fast-forward your system's clock, but you can fast-forward a virtualised clock. It was a design decision in Rx to only use time through schedulers. This decision allows you to replace real time with 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. We decide when time progresses and by how much.
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());Advancing to 1ms
Immediate
Virtual time: 1
Advancing to 10s
Virtual time: 10000
Advancing to 40s
20s
40s
Virtual time: 40000
We scheduled 3 tasks: one to be executed immediately, and two to be executed in the future. Nothing happens until we advance time, including the tasks scheduled immediately. When we advance time, the scheduler synchronously executes all the tasks that were scheduled for that period of time, in the order of the time they were scheduled for.
advanceTimeTo allows you to set time to any value, including one that is before the current time. This implementation decision can needlessly introduce bugs in the tests, so it is probably better to use the next method, when applicable.
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());Advancing by 1ms
Immediate
Virtual time: 1
Advancing by 10s
Virtual time: 10001
Advancing by 40s
20s
40s
Virtual time: 50001
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());Immediate
Virtual time: 0
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);First
Second
Third
Rx operators which involve asynchronous actions schedule those actions using a scheduler. If you take a look at all the operators in Observable, you will see that such operators have overloads that take a scheduler. This is the way that you can supplement their real-time schedulers for your TestScheduler.
Here is an example where we will test the output of Observable.interval against what we expect it to emit.
@Test
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 suggested 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 TestSubscriber, which will do that for us. Its event handlers will collect every notification received and make them available for us to inspect. With TestSubscriber our previous test becomes:
@Test
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, to which the TestSubscriber is subscribed, 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.
| Previous | Next |
|---|---|
| Scheduling and threading | Sequences of coincidence |