How to use CompositeDisposable with IAsyncObservable subscriptions?

[colejohnson66]

For our application, we've been abusing async void to execute subscriptions asynchronously when needed. We then dispose of these subscriptions with a CompositeDisposable object when a navigation event occurs. For example:

void OnNavigatedTo(CompositeDisposables disposables)
{
    this.WhenAnyValue(vm => vm.SomeReactiveProperty)
        .Subscribe(async value =>
        {
            await DoSomething(value);
        })
        .DisposeWith(disposables);
}

We see the new System.Reactive.Async is in preview, and we'd like to use it. However, I'm having trouble figuring out how to bridge the gap between the ValueTask<IDisposable> (returned from SubscribeAsync) and DisposeWith. When does this "task" complete? How do I cancel the subscription through it?

Can anyone provide some guidance here?

[idg10]

When does this "task" complete?

That's up to the observable source.

For example, if you call AsyncObservable.Create, you get to pass in an asynchronous callback. That callback will be invoked when someone calls SubscribeAsync on the IAsyncObservable<T> returned by Create, and SubscribeAsync can't complete until that callback completes.

For example:

IAsyncObservable<int> slow = AsyncObservable.Create(
    async (IAsyncObserver<int> obs)
    {
        // We can take as long as we like...
        await Task.Delay(5000);
        return AsyncDisposable.Nop;
    });

If you call slow.SubscribeAsync(obs), it's just going to call that callback. (In fact the current implementation literally returns the exact ValueTask<IAsyncDisposable> returned by the callback passed to Create.)

So the answer to the question "when does this task complete?" is: "That's outside of Async.Rx's control."

I think (but I'm not 100% sure because we inherited this code, and I've not yet had time to look at every single line of code in Rx.Async) that as long as no external code causes subscription to block, it will complete immediately. As far as I know, all operators just pass the call down to their source and will block only if their source's SubscribeAsync blocks.

I think DisposeWith is a ReactiveUI thing. Although it appears in lots of examples, I've been unable to find documentation for it, but if I've understood correctly, it essentially adds any IDisposable to a CompositeDisposable's list of items to dispose.

So essentially what you want is something which waits for the ValueTask<IDisposable> to complete, and when it does, add that to the CompositeDisposable. So maybe something along these lines:

public static async void DisposeWithAsync(this ValueTask<IDisposable> dt, CompositeDisposable compositeDisposable)
        where T : IDisposable
    {
        compositeDisposable.ArgumentNullExceptionThrowIfNull(nameof(compositeDisposable));

        compositeDisposable.Add(await dt);
    }

However, this is nasty because it's async void. If the SubscribeAsync fails for any reason, and if it does so asynchronously, you'll lose the exception.

To avoid that you'd need some mechanism for handling errors that occur during asynchronous subscription. You can't just block, because the fact that you're using Rx.Async means you're in a world of async. So you would need to make an application-level decision about how you actually want to handle that. What would you want to do if a subscription like this fails, say, several seconds after the call to OnNavigatedTo?

If you're expecting the subscription to be synchronous in practice, you could inspect the ValueTask<IDisposable> to see if it did in fact complete immediately, and throw if it didn't. (In cases where you know that your source has absolutely no reason to cause subscription to be an asynchronous operation, which should definitely be the case if your IAsyncObservable<T> is just a wrapper for an underlying IObservable<T>, then it would actually be valid to effectively assert that you require subscription to be synchronous.)

[colejohnson66]

Thanks for the response! You are correct about DisposeWith; It adds the disposable to a CompositeDisposable we dispose on page change. Additionally, we don't expect any subscriptions to fail, as we only ever (currently) use this.WhenAnyValue from ReactiveUI. Unfortunately, ReactiveUI's codebase is extremely difficult to grok due to interfaces/abstracts at every corner. this.WhenAnyValue always completes synchronously, so perhaps SubscribeAsync always completes synchronously, but I haven't done enough testing.