How do I best subscribe to property changes with an async method?

[SittenSpynne]

In my program I have a ReactiveProperty representing the selected item in a ListView. When that item changes, I need to execute some commands to update some connected hardware. Those commands are async since they have to block and wait for a response from the device.

If I do this, it works, but I don't like it because async void can cause a lot of issues:

class Example
{

    public ReactiveProperty<Item> SelectedItem { get; set; }

    public Example()
    {
        SelectedItem = new ReactiveProperty<Item>();

        SelectedItem.Subscribe(WhenItemChanges).AddTo(Disposables);
    }

    private async void WhenItemChanges(Item item)
    {
        await Something(item);
    }
}

It seems like in general, this is an accepted pattern for doing that with observables:

class Example
{
    public ReactiveProperty<Item> SelectedItem { get; set; }

    public Example()
    {
        SelectedItem = new ReactiveProperty<Item>();

        SelectedItem
            .SelectMany(async (item) => await WhenItemChangesAsync(item))
            .Subscribe()
            .AddTo(Disposables);
    }

    private async Task<Unit> WhenItemChanges(Item item)
    {
        await Something(item);
        return Unit.Default;
    }
}

However, when I do this, nothing happens when the item changes. I know I've seen in the past using Do() that a ReactiveProperty doesn't seem to actually publish its values. I can demonstrate this by being wordier:

SelectedItem
    .Do(x => _logger.Debug("Item changed."))
    .SelectMany(async (item) => await WhenItemChangesAsync(item))
    .Do(x => _logger.Debug("SelectMany emitted"))
    .Subscribe(x => _logger.Debug("Subscription"))
    .AddTo(Disposables);

I get nothing.

So what's the proper way to run an asynchronous method in response to a property changing, without using async void lambdas?

[runceel]

Hi @SittenSpynne,

I think your approach works fine. I have checked that with following code:

using System;
using System.Reactive;
using System.Reactive.Linq;
using System.Threading.Tasks;
using Reactive.Bindings;
using System.Reactive.Disposables;
using Reactive.Bindings.Extensions;

namespace ReactivePropertyTest
{
    class Item 
    {
        private readonly Guid _id = Guid.NewGuid();
        public override string ToString() => _id.ToString();
    }
    class Example : IDisposable
    {
        private CompositeDisposable Disposables { get; } = new CompositeDisposable();
        public ReactiveProperty<Item> SelectedItem { get; }

        public Example()
        {
            SelectedItem = new ReactiveProperty<Item>();
            SelectedItem
                .Do(x => Console.WriteLine($"item changed: {x}"))
                .SelectMany(async (item) => await WhenItemChangesAsync(item))
                .Do(x => Console.WriteLine($"select many emitted"))
                .Subscribe()
                .AddTo(Disposables);
        }

        private async Task<Unit> WhenItemChangesAsync(Item item)
        {
            Console.WriteLine($"{nameof(WhenItemChangesAsync)} started with {item}");
            await Task.Delay(3000);
            Console.WriteLine($"{nameof(WhenItemChangesAsync)} finished with {item}");
            return Unit.Default;
        }

        public void Dispose() => Disposables.Dispose();
    }
    public class Program
    {
        public static void Main(string[] args)
        {
            var ex = new Example();
            ex.SelectedItem.Value = new Item();
            ex.SelectedItem.Value = new Item();

            Console.ReadKey();
            ex.Dispose();
        }
    }
}

The result was:

item changed: 
WhenItemChangesAsync started with 
item changed: 1932dada-ed24-4307-9790-df6590e843c5
WhenItemChangesAsync started with 1932dada-ed24-4307-9790-df6590e843c5
item changed: d9f6a6c7-c072-459d-a2aa-4f9b7b386a1c
WhenItemChangesAsync started with d9f6a6c7-c072-459d-a2aa-4f9b7b386a1c
WhenItemChangesAsync finished with 
WhenItemChangesAsync finished with 1932dada-ed24-4307-9790-df6590e843c5
WhenItemChangesAsync finished with d9f6a6c7-c072-459d-a2aa-4f9b7b386a1c
select many emitted
select many emitted
select many emitted

As common mistakes, you forget .Value or Mode=TwoWay in XAML. Like below:

<!-- valid -->
<ListBox SelectedItem="{Binding SelectedItem.Value, Mode=TwoWay}" />

<!-- invalid -->
<ListBox SelectedItem="{Binding SelectedItem.Value}" />
<ListBox SelectedItem="{Binding SelectedItem, Mode=TwoWay}" />

Could you check your XAML? If you provided minimum Visual Studio project for reproducing, then I will check it well.

[runceel]

I have just thought... Do you care a null value in the WhenItemChangesAsync method? Because when you subscribe ReactiveProperty, in default behavior, ReactiveProperty publishes current value.

Please change your code to below to log exceptions:

SelectedItem
    .Do(x => _logger.Debug("Item changed."))
    .SelectMany(async (item) => await WhenItemChangesAsync(item))
    .Do(x => _logger.Debug("SelectMany emitted"), ex => logger.Debug(ex.ToString()), () => logger.Debug("Completed"))
    .Subscribe(x => _logger.Debug("Subscription"))
    .AddTo(Disposables);

I guess you might get stack trace. If so, your sequence was exit with OnError, when you subscribed it.

[runceel]

Ping to @SittenSpynne

[SittenSpynne]

Sorry for the slow reply, I got swamped and didn't check back in on this.

I see your example does produce the results you say. The example I posted was based on some stuff that was failing for me in a project, but now it's long enough ago I can't remember exactly what it was. The comment about the null item seems like something I might not have considered.

Anyway, now that you've verified this should work, if I get in the situation where it doesn't work again and try the stuff in this answer and it still doesn't work, I'll post a better repro case so we can figure out if I'm still missing something.