RX and INotifyPropertyChanged

By Matthew Adams Co-Founder 2 November 2010

Richard Szalay has a great blog post which allows you to use the Reactive Extensions IObservable pattern to subscribe to INotifyPropertyChanged implementers.

However, the syntax is still a little bit clumsy.

By adding a few extra helpers to the class we can go from this:

viewModel.GetPropertyChangeValues(x => x.TheProperty).Subscribe(GetToWork);

To this

viewModel.Subscribe(x => x.SomeProperty, SomePropertyChanged);

Here's my slightly modified code.

public static class NotifyPropertyChangeReactiveExtensions
{
    // Returns the values of property (an Expression) as they
    // change, starting with the current value
    public static IObservable<TValue>
        GetPropertyValues<TSource, TValue>(
            this TSource source,
            Expression<Func<TSource, TValue>> property)
            where TSource : INotifyPropertyChanged
    {
        MemberExpression memberExpression =
            property.Body as MemberExpression;

        if (memberExpression == null)
        {
            throw new ArgumentException(
                "property must directly access " +
                "a property of the source");
        }

        string propertyName = memberExpression.Member.Name;

        Func<TSource, TValue> accessor = property.Compile();

        return source.GetPropertyChangedEvents()
            .Where(x => x.EventArgs.PropertyName == propertyName)
            .Select(x => accessor(source))
            .StartWith(accessor(source));
    }

    // This is a wrapper around FromEvent(PropertyChanged)
    public static IObservable<IEvent<PropertyChangedEventArgs>>
        GetPropertyChangedEvents(this INotifyPropertyChanged source)
    {
        return Observable.FromEvent<PropertyChangedEventHandler,
            PropertyChangedEventArgs>(
            h => new PropertyChangedEventHandler(h),
            h => source.PropertyChanged += h,
            h => source.PropertyChanged -= h);
    }

    public static IDisposable
        Subscribe<TSource, TValue>(
            this TSource source,
            Expression<Func<TSource, TValue>> property,
            Action<TValue> observer)
        where TSource : INotifyPropertyChanged
    {
        return source
                .GetPropertyValues(property)
                .Subscribe(observer);
    }

    public static IDisposable
        Subscribe<TSource, TValue>(
            this TSource source,
            Expression<Func<TSource, TValue>> property,
            Action<TValue> observer,
            Action onCompleted)
        where TSource : INotifyPropertyChanged
    {
        return source
                .GetPropertyValues(property)
                .Subscribe(observer, onCompleted);
    }

    public static IDisposable
        Subscribe<TSource, TValue>(
            this TSource source,
            Expression<Func<TSource, TValue>> property,
            Action<TValue> observer,
            Action<Exception> onException)
        where TSource : INotifyPropertyChanged
    {
        return source
                .GetPropertyValues(property)
                .Subscribe(observer, onException);
    }

    public static IDisposable
        Subscribe<TSource, TValue>(
            this TSource source,
            Expression<Func<TSource, TValue>> property,
            Action<TValue> observer,
            Action<Exception> onException,
            Action onCompleted)
        where TSource : INotifyPropertyChanged
    {
        return source
                .GetPropertyValues(property)
                .Subscribe(observer, onException, onCompleted);
    }
}

Why should you layer your APIs like this? Read on here...

Matthew Adams on Twitter