Posted in:

In my last post, I compared the code to raise and subscribe to regular .NET events with the Reactive Extensions equivalent. In this post, I want to look at a few reasons you might want to make the transition to Rx Observables.

Idiomatic unsubscribe

Unsubscribing from regular .NET events is straightforward, so long as you don’t use the lambda syntax:

// subscribe:
raiser.Progress += OnProgress;
// unsubscribe:
raiser.Progress -= OnProgress;

// subscribe with lambda:
raiser.Progress += (sender, args) => Console.WriteLine("Progress");
// ... but how do we unsubscribe?

Rx is much nicer. Calling Subscribe returns a disposable subscription object, which you can use to unsubscribe when you’re no longer interested in receiving any more events. This makes it easier to write code that avoids some of the memory leak gotchas that you occasionally can run into with events.

var subscription = raiser.Progress.Subscribe(
 p => Console.WriteLine("Progress {0}", p.Value);
// later if required:
subscription.Dispose()

Also, as Dennis Daume kindly pointed out on Twitter, Rx things even safer by automatically unsubscribing all subscribers once a sequence has completed or errored. So in most cases, you won’t need to unsubscribe at all.

Built-in Errors and Complete

IObservables also have built-in capability to report that the sequence has ended, either successfully (Complete) or with an error. This saves having to create additional events to report problems, or add exception properties to event args that developers need to remember to check.

Filter and Transform

This is where things get really interesting. Observable sequences can be filtered and transformed in a very similar way to enumerable sequences with LINQ. In fact, many of the same operators are available. So I can filter out every other progress event from our last example like this:

eventDemo.Progress
    .Where(p => p.Value % 2 == 0)
    .Subscribe(p => Console.WriteLine("Progress {0}", p.Value));

and I can even move the string formatting into a select statement like this, so I get an observable sequence of strings instead of a sequence of Progress events:

eventDemo.Progress
    .Where(p => p.Value%2 == 0)
    .Select(p => String.Format("Progress {0}", p.Value))
    .Subscribe(Console.WriteLine);

As you can see, IObservable events are composable, and the operators can be chained together. This opens up lots of really cool and interesting possibilities. For example you can throttle events if there are too many and you are only interested in receiving periodic updates:

eventDemo.Progress
    .Throttle(TimeSpan.FromSeconds(3))
    .Select(p => String.Format("Progress {0}", p.Value))
    .Subscribe(Console.WriteLine);

Threading Helpers

One issue you regularly face with events in .NET is that the can be raised on a different thread to the one you need to handle them on. This is an issue when the handler needs to make GUI changes. Rx solves this problem by letting you specify a synchronization context for the events to be handled on:

eventDemo.Progress
    .Select(p => String.Format("Progress {0}", p.Value))
    .ObserveOn(SynchronizationContext.Current)
    .Subscribe(Console.WriteLine);

Observable Timers

Observable also provides a convenient Timer function, which allows you to receive a single event in a specified timespan, or a repeated sequence of events. If can even be given a DateTimeOffset to alert you at a specific time.

var period = TimeSpan.FromMilliseconds(500);
Observable.Timer(period).Subscribe(t => Console.WriteLine("One-off: {0}", t));
Observable.Timer(period, period).Subscribe(t => Console.WriteLine("Repeating: {0}", t));

Observe Regular .NET Events

Obviously regular .NET events aren’t going to go away overnight, but if you like you can convert them into IObservables, allowing you to make use of all the cool stuff that Rx provides. It does require a rather strange syntax, but that’s due to the way .NET events work:

Observable.FromEventPattern<ProgressEventArgs>
    (h => eventDemo.Progress += h, 
     h => eventDemo.Progress -= h)
   .Subscribe(p => Console.WriteLine("Progress {0}",p);

Summary

I’ve barely scratched the surface of what’s possible with Rx and Observables, but as you can see, it’s much more powerful than regular .NET events. It would be nice if everyone using .NET would migrate over to using Observables, but I suspect that won’t happen since although the IObservable interface is now part of the .NET framework, to get the most out of it, you also need the Reactive Extensions libraries. But certainly my experimentations with Rx have made me want to make use of it more frequently rather than just defaulting to plain old .NET events for everything.