The Reactive Extensions (Rx) library is one of those tools that opens up your mind to new ways of thinking about your how your data and application work together. In this post I’m going to talk about an easy technique you can use to replay time series data as if it were live.

Replaying time with HistoricalScheduler

Schedulers in Rx are a deep subject, but one of the things schedulers are responsible for is managing how time based operations run. All time based operations in Rx, like Timeout, Delay, etc allow you to provide your own scheduler.

HistoricalScheduler inherits from VirtualTimeScheduler, and that name gives us an idea what we can do with it. VirtualTimeScheduler gives us control of time – at least as far as our code goes. You can’t go back in time and buy stocks to become rich, but you can make your code think it’s doing that.

HistoricalScheduler gives us a simple interface to replay past events. You can set the scheduler to any time, past or future, and then step forward by increments of time, or run all scheduled events at once. Using three simple methods, we can replay time series data, at any speed we like:

• advanceTo(time) moves the scheduler forward to a specific point in time
• advanceBy(time interval) moves the scheduler forward by a specific amount of time
• start() causes the scheduler to move forward until and only stop once all scheduled events have run

An example – replaying a tweet stream

I ran an Rx Workshop for the Women Who Code Sydney group, and I wanted to pick some data for the workshop that would be more interesting than stock tickers or mouse movements. Twitter seemed like a good idea. The only topic big enough at the time to generate a large stream was the Cricket World Cup, which is not the most fascinating of topics. But it gave me a lot of tweets, which was what I wanted.

I wanted to be able to replay the tweet stream as if it was happening live. Every tweet has a timestamp, so it was easy to schedule each tweet to happen at the time it originally happened. Then using the HistoricalScheduler, I could start time from the first tweet, and replay the stream from there:

var startTime = tweets[0].timestamp;
var scheduler = new Rx.HistoricalScheduler();
scheduler.advanceTo(startTime);

var combinedTweets = Rx.Observable.for(tweets,
function (t) {
    return Rx.Observable.timer(t.timestamp.toDate(), scheduler)
             .map(function () { return t; });
});

This gives us a stream of tweets that will replay at the same rate as they originally did. Notice how the scheduler is used in the call to Observable.timer, otherwise the timer will run off real time, not our virtual time.

Speeding things up

For the workshop, we were doing things like working out how many tweets per minute there were. Being able to replay the tweets faster than normal time meant people could quickly get a good idea of how their time window operations were working. And it’s kind of cool to see just how easy it is to change the speed of the stream.

I mentioned the advanceBy() method on the HistoricalScheduler. If we want to be slow down or speed up time, we just have to translate an interval of real time into a different amount of virtual time on the scheduler:

Rx.Observable.interval(refreshInterval)
  .subscribe(function () {
    scheduler.advanceBy(refreshInterval * timeMultiplier);
  });

Now we can modify the timeMultiplier any way we want to, at any time, and the tweet stream will change its replay speed. And if you change the multiplier to 0, then your stream pauses. Then all you need is a slider to alter the multiplier, and we’re done.

Alternately, you could simply run all tweets in the queue at once, by running scheduler.start(). This would let you see all the output from your logic that would occur if you waited for the tweet stream to complete – but it would only take a few seconds to run. For example, if you had some logic to detect unusual events, such as cases where the rate of tweets suddenly doubles, then being able to instantly evaluate hours worth of data is very powerful.

Other uses for HistoricalScheduler

It’s often the case that your best real-time data comes from production, but you can’t develop new code while listening to that production feed. If you can get your hands on some logs, however, you can create a great testing scenario. You can evaluate new logic and get quick feedback, even if you’re running it against a time series that runs over hours or days. Here are some examples of other uses for this approach.

Testing an alerts system

If you were writing an alerting system that looks for certain events or combinations of events in logs (perhaps errors, or large customer transactions), you could use replay existing logs to evaluate how your code works. You could run days or weeks of logs quickly and check what alerts were raised.

Worst case load testing

You could do worst case load testing on your application using realistic data. You could take logs of your system’s usage from production, and then replay the requests into a test environment at 10 times the rate.

Replaying saved games

I’ve even used a similar approach in a hobby project – a crude recreation of the Missile Command game. By recording all the missile launches in the game, an entire game can be replayed (at twice the speed), so you can watch the games from the highscores list.

Wrapping up

Rx is a great engine for time based application logic. Using the HistoricalScheduler can give you a test-bed that runs just like production – or even better.

For more info, Lee Campbell has a good description of schedulers in his free book, and Bart de Smet has a very in-depth article on using schedulers for testing, including discussion of replaying events.