WhenAnyValue - how to ensure the order of async subscriber?

[sinatrocious]

I've following code:

SomeReactive.WhenAnyValue(o => o.SomeBool).Subscribe(async someBool =>
{
    if(someBool)
    {
        await ...
        Start();
    }
    else
    {
        await ...
        Stop();
    }
});

If SomeBool changes rapidly (and in different threads), then it may lead to a wrong order of calls:

Start()
Stop()
Stop()
Start()

Is there a way to "prevent next call of subscriber, until previous is finished"? I can use lock inside subscriber, but I was hoping for some kind of synchronization from reactive UI side. Is there anything I miss? Or perhaps I need to use another approach?

P.S.: I asked this question on StackOverflow earlier.

[ChrisPulman]

If I was trying to do this I would use Synchronize as follows:

var locker = new object();
SomeReactive.WhenAnyValue(o => o.SomeBool)
	.Synchronize(locker)
	.Subscribe(async someBool =>
	{
	    if(someBool)
	    {
	        await ...
	        Start();
	    }
	    else
	    {
	        await ...
	        Stop();
	    }
	});

Hopefully this works in your situation.

[glennawatson]

Potentially you could use ObserveOn with a thread scheduler

[sinatrocious]

@glennawatson, I doubt it. Maybe I don't understand something fully, but await inside subscriber is my problem: thread is released, so scheduler will continue processing next change (talking about CurrentThreadScheduler). Unless you mean writing some custom scheduler, which is capable of figuring out that subsciber is async and await for it?

[anaisbetts]

No, the solution to this problem is to get your source of asynchrony (the await function) inside the Pipeline - when you write your async method inside the Subscribe, that's a void async, it's completely outside of Rx. Instead, you need to put your async functions inside a SelectMany or a Select.Defer.Concat and allow Rx to constrain the asynchrony. That way it handles all the locking and reentrancy (what you're experiencing) for you

[sinatrocious]

@anaisbetts, async void is the problem indeed. I've seen your answer in SO topic, where Stephen Cleary explains it ;)

I will use SubscribeAsync in place of Subscribe:

public static IDisposable SubscribeAsync<T>(this IObservable<T> source, Func<T, Task> func) =>
            source.Select(o => Observable.FromAsync(_ => func(o)))
                .Concat()
                .Subscribe();