Alternative to using async () => in Rx Finally

user_4140832 0
Viewed 1734

My understanding is that async void, should be avoided and that async () => is just async void in disguise when used with Action.

Hence, using the Rx.NET Finally operator asynchronously with async () => should be avoided since Finally accepts Action as parameter:

IObservable<T>.Finally(async () =>
{
    await SomeCleanUpCodeAsync();
};

However, if this is bad practise, what is then best practice to use in the case where I for instance need to asynchronously close a network connection on OnCompleted or if my observable end with OnError?

async-await system.reactive asynchronous c#
4 Answers

Quoting from the Intro to Rx:

The Finally extension method accepts an Action as a parameter. This Action will be invoked if the sequence terminates normally or erroneously, or if the subscription is disposed of.

(emphasis added)

This behavior cannot be replicated by a Finally operator that accepts a Func<Task> parameter, because of how the IObservable<T> interface is defined. Unsubscribing from an observable sequence is achieved by calling the Dispose method of the IDisposable subscription. This method is synchronous. And the whole Rx library is built on top of this interface. So even if you create an extension method DisposeAsync for IDisposables, the built-in Rx operators (for example Select, SelectMany, Where, Take etc) will be unaware of its existence, and will not invoke it when they unsubscribe from their source sequence. A subscription chain of operators will be automatically unlinked by calling the synchronous Dispose method of the previous link as always.

Btw there has been an attempt to implement an asynchronous version of Rx (AsyncRx), that is built on top of the completely new interfaces that are shown below. This library has not been released yet.

public interface IAsyncObserver<in T>
{
    ValueTask OnNextAsync(T value);
    ValueTask OnErrorAsync(Exception error);
    ValueTask OnCompletedAsync();
}

public interface IAsyncObservable<out T>
{
    ValueTask<IAsyncDisposable> SubscribeAsync(IAsyncObserver<T> observer);
}

public interface IAsyncDisposable
{
    public ValueTask DisposeAsync();
}
user_111785490
Related