Alternative to using async () => in Rx Finally

Question

I understand that async void, should be avoided and that async () =>just async voidhides when used with Action.

Therefore, you should avoid using the AIX Rx.NET operator asynchronously with async () =>, since Finally it takes Actionas a parameter:

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

However, if this is bad practice, then it is best to use when I need to close the network agreement asynchronously , if my observed value is OnError or OnCompleted?

Answer 1

, Async void async () => async void .

. async () => Func<Task> (), Action (). / , , async void, , async Task .

AsyncFinally, Func<Task> Action, Observable.Finally:

public static class X
{
    public static IObservable<T> AsyncFinally<T>(this IObservable<T> source, Func<Task> action)
    {
        return source
            .Materialize()
            .SelectMany(async n =>
            {
                switch (n.Kind)
                {
                    case NotificationKind.OnCompleted:
                    case NotificationKind.OnError:
                        await action();
                        return n;
                    case NotificationKind.OnNext:
                        return n;
                    default:
                        throw new NotImplementedException();
                }
            })
            .Dematerialize()
        ;
    }
}

:

try
{
    Observable.Interval(TimeSpan.FromMilliseconds(100))
        .Take(10)
        .AsyncFinally(async () =>
        {
            await Task.Delay(1000);
            throw new NotImplementedException();
        })
        .Subscribe(i => Console.WriteLine(i));
}
catch(Exception e)
{
    Console.WriteLine("Exception caught, no problem");
}

AsyncFinally Finally, .

Answer 2

Rx, ; async void, . , , "" Rx.

OnErrorResumeNext() . OnErrorResumeNext() , , , :

var myObservable = ...

myObservable
    .Subscribe( /* Business as usual */ );

Observable.OnErrorResumeNext(
        myObservable.Select(_ => Unit.Default),
        Observable.FromAsync(() => SomeCleanUpCodeAsync()))
    .Subscribe();

myObservable ConnectableObservable (, Publish()), .

Answer 3

Finally

public static IObservable<TSource> Finally<TSource>(
    this IObservable<TSource> source,
    Action finallyAction
)

, .

, - , async void Task.Factory , .