Reactive UI Programming Part 2 – Trail By

This is Part 2 of a series of articles on Reactive UI Programming with Knockout and Reactive Extensions for JavaScript.

Previous parts:

• Introduction
• Part 1 – Converting between KO and RxJs

Recently I was working on a GUI that walked the user through a series of 5 steps in a Workflow. The user would spend most of their time on the 3rd step. As the user made edits, those edits would immediately be sent to the server for processing, and results from the server would show up in a different part of the GUI. There was a requirement that whenever the user left the 3rd step, we needed to send a message to the server to commit the changes they’d made.

So basically whenever the user was on the 3rd step, and then chose to move forward or backward in the workflow, we needed to send a command to the server.

As I started thinking about how to implement this, my first thought was to just subscribe to the currentStep ko.observable property on our view model and issue the command like so:

var previous = undefined;self.currentStep.toRx(true).where(function (newCurrentStep) {var prev = previous;previous = newCurrentStep;return prev === 3;}).selectMany(function () { return self.model().commitStep3ChangesRx().catchException(errorHandler); }).subscribe();

This sets up an Rx query which listens to changes in currentStep. It records the previous value in a temporary variable (previous). Whenever currentStep changes, it checks if the previous step was 3. If so, then it calls a function on our model to update the server, which returns an Rx.Observable which we can use to monitor the results of the command. We then subscribes to this Rx query, which activates it.

This code would certainly do what we want, but I wasn’t very happy with it for a couple of reasons.

First, the intent of the code is a little bit obfuscated by the way we track the previous step. If a developer goes to do some work on this view model, it will take them just a little bit longer to parse this code and understand what it is doing. So it is a little messy.

Second, it isn’t really re-usable. Tracking the previous value of a property seems like a pretty generic concept. It would be nice to capture this concept as a method so that it can be re-used the next time this comes up. This would also solve my first problem.

A word about our models

As you see in the code above, we use our models to communicate with the server. This keeps our view models focused on interacting with the user through the view. All of the commands and queries we send to the server result in an Rx.Observable. We subscribe to these observables so that we can be notified of the results of our call. This is similar in concept to the jQuery Deferred that is returned by jQuery.ajax. We prefer to use Rx.Observables instead of jQuery.Deferred because of the richer RxJs linq library and because we have a number of observable queries, which can return many results over time.

In many cases, our models actually expose queries and commands as simple ko.observable properties. The model listens to an observable query to the server to keep the property value up to date, and when a view model writes to the property, the model issues a command to the server to notify the server of the change. But some commands, like our commitStep3Changes are not really representable as a property. Such commands are exposed as methods on our models which return an Rx.Observable.

trailBy

Once I decided to write a general purpose method that could track prior values of a property, I just needed to decide whether it should be a knockout extension, or a Rx extension. Since I would ultimately be using it to launch an Rx query, and since the extensions syntax of Rx is a bit cleaner than Knockout, I decided to make it an Rx operator. I also decided to generalize it to support more than just the previous value. Like, what if I wanted to know the value 10 changes ago?

The result is the trailBy operator:

Rx.Observable.prototype.trailBy = function (n) {/// <summary>/// Converts the source stream into a new stream where the events are delayed by <c>n</c> events./// For example, source.trailBy(10) will yield the first event when <c>source</c> produces the 11th event./// </summary>/// <param name="n" type="Number">How many events to lag behind. 0 would produce the same stream as source. Must not be negative</param>/// <returns type="Rx.Observable"></returns> var source = this;if (n < 0) { thrownew Error("n must not be negative"); }if (n === 0) { return source; }return Rx.Observable.createWithDisposable(function (observer) {var ringBuffer = new Array(n),i = 0,ready; return source.subscribe(function (value) {if (ready) { observer.onNext(ringBuffer[i]); }ringBuffer[i++] = value;if (i === n) {ready = true;i = 0;}}, observer.onError.bind(observer), observer.onCompleted.bind(observer));});};

After validating the n parameter and short-circuiting the n === 0 case, we return a new Rx.Observable, which when subscribed to will create a ring buffer that is just big enough to store n values. We then subscribe to source and start listening for results. When we have received n values and filled up our buffer, we are ready to start yielding results. Now, as each new item arrives, we send out the oldest item in our ring buffer before storing the new value at that location.

Notice that if source produces an error, we immediately send the error to our observer. This follows Rx Guidelines. Errors should be sent immediately and not held. Also notice we immediately send the completion event when source completes. Once source completes, anything left in our buffer is just thrown away. This is due to the semantics of our trailBy operator. If the resulting stream the “previous” value of source, and source has ended, then it stands to reason that the last value of sourcewill never become the “previous” value.

Closing thoughts

With my new trailBy operator, I was able to satisfy the original requirements with this code in my view model:

self.currentStep.toRx(true).trailBy(1).where(function (previousStep) { return previousStep === 3; }).selectMany(function () { return self.model().commitStep3ChangesRx().catchException(errorHandler); }).subscribe();

This is about half as many lines of code as the original version which makes it much easier for a developer to look at and see what’s going on. We’ve also got a nice new operator in our toolbox ready to be used next time.

Here’s a jsFiddle with the code and a unit test.