RX Framework – Building a message bus


I’ve been toying around with the Reactive Extensions (RX) Framework for .NET 4 the last few days and I think I’ve found a quite nice usecase for it;
Since RX is all about sequences of events/messages, it does fit very well together with any sort of message bus or event broker.

Just check this out:

Our in proc message bus:

public class MiniVan
{
    private Subject<object> messageSubject = new Subject<object>();

    public void Send<T>(T message)
    {
        messageSubject.OnNext(message);
    }

    public IObservable<T> AsObservable<T>()
    {
        return this
            .messageSubject
            .Where(m => m is T)
            .Select(m => (T)m);
    }
}

Subscribing to messages:

bus.AsObservable<MyMessage>()
    .Do(m => Console.WriteLine(m))
    .Subscribe();

The nice thing about this is that you get automatic Linq support since it is built into RX.
So you can add message handlers that filters or transform messages.
Pretty slick isnt it?

I’m currenty writing an example IRC chat client based on this idea which I will publish in a week or two.

//Roger

6 thoughts on “RX Framework – Building a message bus

  1. I refactored the method using using LINQ.
    public IObservable AsObservable() { return this .messageSubject.OfType();}

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out / Change )

Twitter picture

You are commenting using your Twitter account. Log Out / Change )

Facebook photo

You are commenting using your Facebook account. Log Out / Change )

Google+ photo

You are commenting using your Google+ account. Log Out / Change )

Connecting to %s