Rx INotifyPropertyChanged to IObservable<Tuple<TProperty,TProperty>>

Question

This must be a case of poor googling on my part as I know I saw a solution for this out on the wbe before, but how would I go about implementing an extension method which is able to convert INotifyPropertyChanged.PropertyChanged events into an IObservable<Tuple<TProperty,TProperty>> where the values of the tuple represent the oldValue and the newValue of the property?

So I want to know what is the best way to take something like this: (credit for below to)

public static IObservable<TProperty> ObservePropertyChanged<TNotifier, TProperty>(this TNotifier notifier,
    Expression<Func<TNotifier, TProperty>> propertyAccessor,
    bool startWithCurrent = false)
    where TNotifier : INotifyPropertyChanged {

    // Parse the expression to find the correct property name.
    MemberExpression member = (MemberExpression)propertyAccessor.Body;
    string name = member.Member.Name;

    // Compile the expression so we can run it to read the property value.
    var reader = propertyAccessor.Compile();

    var propertyChanged = Observable.FromEventPattern<PropertyChangedEventHandler, PropertyChangedEventArgs>(
        handler => (sender, args) => handler(sender, args),
        x => notifier.PropertyChanged += x,
        x => notifier.PropertyChanged -= x);

    // Filter the events to the correct property name, then select the value of the property from the notifier.
    var newValues = from p in propertyChanged
                    where p.EventArgs.PropertyName == name
                    select reader(notifier);

    // If the caller wants the current value as well as future ones, use Defer() so that the current value is read when the subscription
    // is added, rather than right now. Otherwise just return the above observable.
    return startWithCurrent ? Observable.Defer(() => Observable.Return(reader(notifier)).Concat(newValues)) : newValues;
}

And convert it to fit this signature:

public static IObservable<Tuple<TProperty,TProperty>> ObservePropertyChanged<TNotifier, TProperty>(this TNotifier notifier,
    Expression<Func<TNotifier, TProperty>> propertyAccessor,
    bool startWithCurrent = false)
    where TNotifier : INotifyPropertyChanged {

    // Parse the expression to find the correct property name.
    MemberExpression member = (MemberExpression)propertyAccessor.Body;
    string name = member.Member.Name;

    // Compile the expression so we can run it to read the property value.
    var reader = propertyAccessor.Compile();

    var propertyChanged = Observable.FromEventPattern<PropertyChangedEventHandler, PropertyChangedEventArgs>(
        handler => (sender, args) => handler(sender, args),
        x => notifier.PropertyChanged += x,
        x => notifier.PropertyChanged -= x);

    // Filter the events to the correct property name, then select the value of the property from the notifier.
    var newValues = from p in propertyChanged
                    where p.EventArgs.PropertyName == name
                    select reader(notifier);

    throw new NotImplementedException();
}

Edit: I figured out something that seems to work after trying many different operators. Is this a correct way of accomplishing this? Is there anything I'm missing?

public static IObservable<Tuple<TProperty,TProperty>> ObserveValueChanged<TNotifier, TProperty>(this TNotifier notifier,
    Expression<Func<TNotifier, TProperty>> propertyAccessor,
    bool startWithCurrent = false)
    where TNotifier : INotifyPropertyChanged {
    var observable = ObservePropertyChanged(notifier, propertyAccessor, startWithCurrent);

    return observable.Scan(new Tuple<TProperty, TProperty>(default(TProperty), default(TProperty)),
                    (acc, p) => new Tuple<TProperty, TProperty>(acc.Item2, p));

}

Edit: I incorporated Gideon's solution to end up with the following:

public static IObservable<Tuple<TProperty, TProperty>> ObserveValueChanged2<TNotifier, TProperty>(this TNotifier notifier,
    Expression<Func<TNotifier, TProperty>> propertyAccessor,
    bool startWithCurrent = false)
    where TNotifier : INotifyPropertyChanged {

    // Compile the expression so we can run it to read the property value.
    var reader = propertyAccessor.Compile();

    var newValues = ObservePropertyChanged(notifier, propertyAccessor, false);
    if (startWithCurrent) {
        var capturedNewValues = newValues; //To prevent warning about modified closure
        newValues = Observable.Defer(() => Observable.Return(reader(notifier))
                                .Concat(capturedNewValues));
    }

    return Observable.Create<Tuple<TProperty, TProperty>>(obs => {
        Tuple<TProperty, TProperty> oldNew = null;
        return newValues.Subscribe(v => {
                if (oldNew == null) {
                    oldNew = Tuple.Create(default(TProperty), v);
                } else {
                    oldNew = Tuple.Create(oldNew.Item2, v);
                    obs.OnNext(oldNew);
                }
            },
            obs.OnError,
            obs.OnCompleted);
    });
}

P.S. I eventually stumbled upon my current solution, but I don't want to violate any etiquite for SO, should I add an answer or close the question (I'd prefer not to delete since this may prove useful later)? I'm still not sure this is the best way of doing this.

Answer 1

If you want to stick to existing operators, Zip along with Skip would probably be the closest to what you need. I would probably write it myself like this (picking up where you throw the NotImplemented):

if (startWithCurrent)
{
    newValues = Observable.Defer(() => Observable.Return(reader(notifier))
                          .Concat(newValues));
}

return Observable.Create<Tuple<TProperty, TProperty>>(obs =>
    {
        Tuple<TProperty, TProperty> oldNew = null;
        return newValues.Subscribe(v =>
            {
                if (oldNew == null)
                {
                    oldNew = Tuple.Create(default(TProperty), v);
                }
                else
                {
                    oldNew = Tuple.Create(oldNew.Item2, v);
                    obs.OnNext(oldNew);
                }
            },
            obs.OnError,
            obs.OnCompleted);
    });