How to Subscribe with async method in Rx?
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.
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);
}