Reacting to Rx

I’ve finally got round to spending a bit of time looking at Rx over the weekend, and my head is still spinning as to just how fantastically relevant this is to some of the stuff I’m working on right now. If have no idea what Rx is, check out these brief Channel 9 videos:

• Erik Meijer: Rx in 15 Minutes - Rx is Here!!!!!
• Writing your first RX Application (6 min)

The first will get you interested, the second will make the penny drop[1].

So anyway, I have a class called a MessagePump<T>. Its job is to abstract away a lot of low-level socket guff (fragmentation, parsing etc…) and just deliver messages as they are read off a socket. It basically just sits in a big async loop of BeginRead / EndRead operations, constantly passing itself as the callback (ie never ‘owning’ a thread).

That’s all it does, so to deliver messages into the rest of the system it exposes a MessageReceived event. And sometimes a message might not parse properly, probably someone got out of sync whatever, so there’s a ExceptionReceived event. Oh, and if you get a zero-byte read from BeginRead that means the socket the other end closed, so there’s a Disconnected event

• MessageReceived(object, EventArgs<T>)
• ExceptionReceived(object, EventArgs<Exception>)
• Disconnected(object, EventArgs)

Now compare this to Rx’s IObserver<T> interface:

• OnNext(T)
• OnError(Exception)
• OnCompleted()

It’s like completely the same. I guess there are only so many ways to skin a cat, but I wasn’t expecting it to be quite so aligned. Hopefully I can read this as saying my design is basically sound.

But whatever, what it really means is that dropping in Rx is going to be a bit of a doddle. In fact because the IObserver<T> and IObservable<T> interfaces (alone) are part of the .net 4 framework, even without Rx I can implement the pattern (just without the Rx fruit),which makes leveraging Rx later on (e.g. to filter with Linq) an option for the consumer.

And because the IObserver<T> / IObservable<T> pattern is much more amenable to composition than a raw .net event (which is really, the whole point of Rx), we can use containers like MEF to attach the subscribers at runtime, with (what seems to be) relative ease.

Both temporal and binary decoupling. Cool.

 

[1] For example: did you ever write something like an auto-complete popup? You want to wait a while after each keystroke in case the user didn’t finish typing yet (about 500ms I think). I ended up writing a general-purpose event-buffer class, that only propagated the event after a specified inactivity period (this also worked great for file change notifications). In Rx this is trivial: just use the ‘Throttle’ linq operator over the event sequence. See the hands-on-lab