Reactive Extensions - Properties update each other

Question

I have 2 DecimalUpDown controls, num_one and num_two, bound to properties First and Second respectively. When First is changed it will contact a server to calculate the value of Second, and vice-versa. Firing the server calls asynchronously freed the UI but, upon quick firing (scroll wheel for example), the last request isn't always the last to return so the values may become out of sync.

Using Reactive I'm trying to Throttle the calls to only fire the server call after the user has stopped making changes for a little while. The problem is that when you make a change during an update, the Properties changing start triggering each other and get stuck in back and forth depending on the TimeSpan of the Throttle.

    public MainWindow()
    {
        InitializeComponent();

        DataContext = this;

        Observable.FromEventPattern<RoutedPropertyChangedEventHandler<object>, RoutedPropertyChangedEventArgs<object>>(h => num_one.ValueChanged += h, h => num_one.ValueChanged -= h)
            .Throttle(TimeSpan.FromMilliseconds(100), Scheduler.ThreadPool)
           .Subscribe(x =>
           {
               Thread.Sleep(300); // simulate work
               Second = (decimal)x.EventArgs.NewValue / 3.0m;
           });

        Observable.FromEventPattern<RoutedPropertyChangedEventHandler<object>, RoutedPropertyChangedEventArgs<object>>(h => num_two.ValueChanged += h, h => num_two.ValueChanged -= h)
            .Throttle(TimeSpan.FromMilliseconds(100), Scheduler.ThreadPool)
           .Subscribe(x =>
           {
               Thread.Sleep(300); // simulate work
               First = (decimal)x.EventArgs.NewValue * 3.0m;
           });
    }

    private decimal first;
    public decimal First
    {
        get { return first; }
        set
        {
            first = value;
            NotifyPropertyChanged("First");
        }
    }

    private decimal second;
    public decimal Second
    {
        get { return second; }
        set
        {
            second = value;
            NotifyPropertyChanged("Second");
        }
    }

score 7 · Accepted Answer · answered Jun 29 '12 at 00:06

There's an inbuilt Rx operator that can help you do exactly what you want without using Throttle and timeouts - it's the Switch operator.

The Switch operator doesn't work on IObservable<T> so most times you would never see it in intellisense.

Instead it operates on IObservable<IObservable<T>> - a stream of observables - and it flattens the source to IObservable<T> by continually switching to the latest observable produced (and ignoring any values from the previous observables). It only completes when the outer observable completes and not the inner ones.

This is exactly what you want - if a new value change occurs then ignore any previous results and only return the latest one.

Here's how to do it.

First I removed the yucky event handling code into a couple of observables.

var ones =
    Observable
        .FromEventPattern<
            RoutedPropertyChangedEventHandler<object>,
            RoutedPropertyChangedEventArgs<object>>(
            h => num_one.ValueChanged += h,
            h => num_one.ValueChanged -= h)
        .Select(ep => (decimal)ep.EventArgs.NewValue);

var twos =
    Observable
        .FromEventPattern<
            RoutedPropertyChangedEventHandler<object>,
            RoutedPropertyChangedEventArgs<object>>(
            h => num_two.ValueChanged += h,
            h => num_two.ValueChanged -= h)
        .Select(ep => (decimal)ep.EventArgs.NewValue);

Your code seems to be a bit muddled. I assume that the value of the DecimalUpDown controls are inputs to the server function that returns the result. So here are the functions that will call the server.

Func<decimal, IObservable<decimal>> one2two = x =>
    Observable.Start(() =>
    {
        Thread.Sleep(300); // simulate work
        return x / 3.0m;
    });

Func<decimal, IObservable<decimal>> two2one = x =>
    Observable.Start(() =>
    {
        Thread.Sleep(300); // simulate work
        return x * 3.0m;
    });

Obviously you put in your actual server code calls in these two functions.

Now it is almost trivial to wire up the final observables and subscriptions.

ones
    .DistinctUntilChanged()
    .Select(x => one2two(x))
    .Switch()
    .Subscribe(x =>
    {
        Second = x;
    });

twos
    .DistinctUntilChanged()
    .Select(x => two2one(x))
    .Switch()
    .Subscribe(x =>
    {
        First = x;
    });

The DistinctUntilChanged makes sure we only make the call if the values actually changed.

Then it's easy to call the two server functions, perform the Switch and get back only the latest result which is then just assigned to the property.

You may need to pop in a scheduler here or there and an ObserveOn to get the subscription over to the UI thread, but otherwise this solution should work nicely.

That's perfect, thank you. Is there any way to throttle the server calls so that lots of requests aren't being sent? I had a go but bumped into the same problem of them 'fighting' each other. — Mim Hufford, Jun 29 '12 at 10:38
Adding a .Sample(xxx) before the .Select(x => serverCall()) seems to do the trick? — Mim Hufford, Jun 29 '12 at 10:52
@MimHufford - Take a look at this answer: http://stackoverflow.com/a/7604825/259769. It has a `BufferWithInactivity` extension method that will buffer with a timeout. Use it with `Switch` and it will give you the best solution. — Enigmativity, Jun 30 '12 at 00:04

score 1 · Answer 2 · answered Jun 28 '12 at 18:48

Property changed notifications should not be fired if the property hasn't changed. You need to add an if statement to the properties.

public decimal First
{
    get { return first; }
    set
    {
        if(first == value)
            return;

        first = value;
        NotifyPropertyChanged("First");
    }
}

Reactive Extensions - Properties update each other

2 Answers2

Linked