How to Subscribe with async method in Rx?

Question

I have following code:

IObservable<Data> _source;

...

_source.Subscribe(StoreToDatabase);

private async Task StoreToDatabase(Data data) {
    await dbstuff(data);
}

However, this does not compile. Is there any way how to observe data asynchronously? I tried async void, it works, but I feel that given solution is not feasible.

I also checked Reactive Extensions Subscribe calling await, but it does not answer my question (I do not care about the SelectMany result.)

Answer 1

You don't have to care about the SelectMany result. The answer is still the same... though you need your task to have a return type (i.e. Task<T>, not Task).

Unit is essentially equivalent to void, so you can use that:

_source.SelectMany(StoreToDatabase).Subscribe();

private async Task<Unit> StoreToDatabase(Data data)
{
    await dbstuff(data);
    return Unit.Default;
}

This SelectMany overload accepts a Func<TSource, Task<TResult> meaning the resulting sequence will not complete until the task is completed.

Answer 2

Late answer, but I think that the following extension methods correctly encapsulate what Charles Mager proposed in his answer:

public static IDisposable SubscribeAsync<T>(this IObservable<T> source, 
                         Func<Task> asyncAction, Action<Exception> handler = null)
{
    Func<T,Task<Unit>> wrapped = async t =>
    {
        await asyncAction();
        return Unit.Default;
    };
    if(handler == null)
        return source.SelectMany(wrapped).Subscribe(_ => { });
    else
        return source.SelectMany(wrapped).Subscribe(_ => { }, handler);
}

public static IDisposable SubscribeAsync<T>(this IObservable<T> source, 
                         Func<T,Task> asyncAction, Action<Exception> handler = null)
{
    Func<T, Task<Unit>> wrapped = async t =>
    {
        await asyncAction(t);
        return Unit.Default;
    };
    if(handler == null)
        return source.SelectMany(wrapped).Subscribe(_ => { });
    else
        return source.SelectMany(wrapped).Subscribe(_ => { }, handler);
}