Is it possible to use Rx "Using" operator with IAsyncDisposable?

Question

Is it possible to use the Using operator in Rx.Net with a resource that implements IAsyncDisposable rather than IDisposable? If not, is there some sort of workaround that I could use?

Answer 1

Here is a Using method that works with IAsyncDisposable objects:

/// <summary>
/// Constructs an observable sequence that depends on a resource object,
/// whose lifetime is tied to the resulting observable sequence's lifetime.
/// </summary>
public static IObservable<TResult> Using<TResult, TResource>(
    Func<TResource> resourceFactory,
    Func<TResource, IObservable<TResult>> observableFactory)
    where TResource : IAsyncDisposable
{
    return Observable.Defer(() =>
    {
        TResource resource = resourceFactory();
        IObservable<TResult> observable;
        try { observable = observableFactory(resource); }
        catch (Exception ex) { observable = Observable.Throw<TResult>(ex); }

        Lazy<Task> lazyDisposeTask = new(() => resource.DisposeAsync().AsTask());
        IObservable<TResult> disposer = Observable
            .FromAsync(() => lazyDisposeTask.Value)
            .Select(_ => default(TResult))
            .IgnoreElements();

        return observable
            .Catch((Exception ex) => disposer.Concat(Observable.Throw<TResult>(ex)))
            .Concat(disposer)
            .Finally(() => lazyDisposeTask.Value.GetAwaiter().GetResult());
    });
}

This method has identical signature with the Rx Observable.Using method (apart from the where clause), and it can be used in the same way.

This implementation takes care of all completion cases:

1. Successful completion: The IAsyncDisposable resource is disposed asynchronously by the Concat operator.
2. Completion with error: The IAsyncDisposable resource is disposed asynchronously by the Catch operator.
3. The sequence is unsubscribed before its completion: The IAsyncDisposable resource is disposed synchronously by the Finally operator. Disposing asynchronously the resource is not possible in this case, for reasons explained here.

Variant with asynchronous factory methods:

public static IObservable<TResult> Using<TResult, TResource>(
    Func<CancellationToken, Task<TResource>> resourceFactoryAsync,
    Func<TResource, CancellationToken, Task<IObservable<TResult>>> observableFactoryAsync)
    where TResource : IAsyncDisposable
{
    return Observable.Create<TResult>(async (observer, cancellationToken) =>
    {
        TResource resource = await resourceFactoryAsync(cancellationToken);
        IObservable<TResult> observable;
        try { observable = await observableFactoryAsync(resource, cancellationToken); }
        catch { await resource.DisposeAsync(); throw; }

        Lazy<Task> lazyDisposeTask = new(() => resource.DisposeAsync().AsTask());
        IObservable<TResult> disposer = Observable
            .FromAsync(() => lazyDisposeTask.Value)
            .Select(_ => default(TResult))
            .IgnoreElements();

        return observable
            .Catch((Exception ex) => disposer.Concat(Observable.Throw<TResult>(ex)))
            .Concat(disposer)
            .Finally(() => lazyDisposeTask.Value.GetAwaiter().GetResult())
            .Subscribe(observer);
    });
}