Adding Rx Framework to Courier

In a previous series of posts I introduced the Courier framework. Recently I have been doing some refactoring on the project and feature enhancements. The refactoring was fairly straight forward.

I was never quite happy with the way that you unregistered for a message so I changed the model of message registration to return a token and then subsequently changed the signature of unregister to take in this token. That change was pretty straight forward and badly needed.

The big feature in this update is the integration with the Rx Framework. If you don’t understand Rx I suggest diving into these videos. Be forewarned, the Rx guys are incredibly smart and tend to blaze through things very quickly. Much is left “up to the reader” to figure out and play with on their own. This makes it pretty difficult to grok. I am not sure I really understand all the moving parts of Rx.

One of my main goals in adding Rx to the Courier project was to allow you do include or exclude the Rx features with minimal pain. I also wanted to avoid a deep dependency on the Rx extension assembly. So I created a separate assembly for the Rx features that introduces an extension method to the Messenger class. This gives you an additional way to subscribe to a message and get an IObservable back. Since it is in a separate assembly you should be able to omit the assembly (and any dependency on Rx) pretty easily if you wanted to.

So what does this change let you do? Essentially I have added one extension method to the messenger that gives you different method to subscribe for a message. Syntactically it looks like this

IObservable myObservable = MyMessenger.RegisterForMessage("Foo");

This tells the messenger that you want to listen for the “Foo” message and get the result pushed into your myObservable sequence. So why is this cool? LINQ.

Plain and simple IObservable is queryable with LINQ and that makes for very interesting scenarios when you are dealing with a message pump that is very active and you want to filter out messages based on a criteria. I have added an example of how to use the new Rx extension method and tried to model the scenario into a real world situation.

Imagine you are listening to WCF service that is sending messages from a Blood Pressure monitor. Imagine the Messenger is broadcasting messages each time the Blood Pressure monitor sends some data.   This can lead to a flood of messages firing through the Messenger. Using this new extension method you can write a simple LINQ query to filter these messages and only “react” to the ones you care about.

In the scenario I show in the sample project I only care about messages when the blood pressure is outside of the “normal” range. In those cases I want to “react” and show some alert. I have written a quick and dirty simulation that will fire messages every second and randomize the pressure. When the pressure is outside of the predefined range I will get a message “pushed” into my IObservable sequence. When this happens I have a delegate that “reacts” to this “push” and does something (in my case I raise an alert)

So, from the snippet above I can take it one step further and do this:

IObservable<Object> observable = from msg in Mediator.RegisterForMessage("BloodPressureMessage")
                                 where (((BloodPressure)msg).Systolic < 90 || ((BloodPressure)msg).Systolic > 119)
                                 && (((BloodPressure)msg).Diastolic < 60 || ((BloodPressure)msg).Diastolic > 79)
                                 select msg;

observable.Subscribe(OnAlertReceived);

Notice how I am now writing a LINQ query on the IObservable that the messenger is returning from the RegisterForMessage method. The line after the LINQ query is how I subscribe to the “push” events that happen on the observable sequence. OnAlertReceived is the delegate I want invoked when a new message is “pushed” on the IObservavle sequence.

Cool? I think so. Also, this check in I am going to promote to Beta 2 status. I have added unit tests and done some other small cleanup work. I think it is getting close to being ready for a true release. I would like to buff out the sample application a bit more and get some more API documentation in place and work out and remaining bugs.

one little nit that I want to clean up in the next version is the casting of the “msg” in the LINQ Query. There are some internal issues with how I am implementing Rx under the covers that makes strong typing of the resulting observable challenging, I don’t have a solution for this yet given my current implementation, so if you find one please contribute it back to the project or let me know.