How to call back async function from Rx Subscribe?

Question

I would like to call back an async function within an Rx subscription.

E.g. like that:

public class Consumer
{
    private readonly Service _service = new Service();

    public ReplaySubject<string> Results = new ReplaySubject<string>();

    public void Trigger()
    {
        Observable.Timer(TimeSpan.FromMilliseconds(100)).Subscribe(async _ => await RunAsync());
    }

    public Task RunAsync()
    {
        return _service.DoAsync();
    }
}

public class Service
{
    public async Task<string> DoAsync()
    {
        return await Task.Run(() => Do());
    }

    private static string Do()
    {
        Thread.Sleep(TimeSpan.FromMilliseconds(200));
        throw new ArgumentException("invalid!");
        return "foobar";
    }
}

[Test]
public async Task Test()
{
    var sut = new Consumer();
    sut.Trigger();
    var result = await sut.Results.FirstAsync();
}

What needs to be done, in order to catch the exception properly?

I just figured that I can put async at the first place. But unfortunately this doesn't solve the problem I am faced with. I will post a more explicit example. — Martin Komischke, Apr 11 '14 at 08:38
possible duplicate of [Is there a way to subscribe an observer as async](http://stackoverflow.com/questions/18814805/is-there-a-way-to-subscribe-an-observer-as-async) — Zache, Apr 11 '14 at 08:48

score 39 · Answer 1 · edited Mar 01 '21 at 18:46

39

Ana Betts' answer works in most scenarios, but if you want to block the stream while waiting for the async function to finish you need something like this:

Observable.Interval(TimeSpan.FromSeconds(1))
          .Select(l => Observable.FromAsync(asyncMethod))
          .Concat()
          .Subscribe();

Or:

Observable.Interval(TimeSpan.FromSeconds(1))
          .Select(_ => Observable.Defer(() => asyncMethod().ToObservable()))
          .Concat()
          .Subscribe();

edited Mar 01 '21 at 18:46

TeaDrivenDev

6,576
32
50

answered May 04 '15 at 12:45

reijerh

732
1
7
15

1

It's another use case but very interesting though. – Martin Komischke Mar 04 '16 at 11:08

score 21 · Answer 2 · answered Apr 11 '14 at 16:45

21

Change this to:

Observable.Timer(TimeSpan.FromMilliseconds(100))
    .SelectMany(async _ => await RunAsync())
    .Subscribe();

Subscribe doesn't keep the async operation inside the Observable.

answered Apr 11 '14 at 16:45

Ana Betts

72,589
16
137
204

2

Thanks Paul, for your suggestion. Very interesting. Do you have something where I can read a little bit further about that? – Martin Komischke May 05 '14 at 07:25
1

Note for others using this solution, if you need keep the order just replace `SelectMany` with `Select` and then `.Concat` (similar to reijerh's answer). – bokibeg Oct 28 '19 at 19:53

score 18 · Accepted Answer · answered Apr 11 '14 at 11:15

You don't want to pass an async method to Subscribe, because that will create an async void method. Do your best to avoid async void.

In your case, I think what you want is to call the async method for each element of the sequence and then cache all the results. In that case, use SelectMany to call the async method for each element, and Replay to cache (plus a Connect to get the ball rolling):

public class Consumer
{
    private readonly Service _service = new Service();

    public IObservable<string> Trigger()
    {
        var connectible = Observable.Timer(TimeSpan.FromMilliseconds(100))
            .SelectMany(_ => RunAsync())
            .Replay();
        connectible.Connect();
        return connectible;
    }

    public Task<string> RunAsync()
    {
        return _service.DoAsync();
    }
}

I changed the Results property to be returned from the Trigger method instead, which I think makes more sense, so the test now looks like:

[Test]
public async Task Test()
{
    var sut = new Consumer();
    var results = sut.Trigger();
    var result = await results.FirstAsync();
}

What if you want to catch exceptions thrown from RunAsync? What I got is: obs.SelectMany(x => Observable.FromAsync(() => RunAsync()).Catch(Observable.Empty())) Is there a better way? — Victor Grigoriu, Apr 01 '15 at 14:59
@VictorGrigoriu: I recommend you ask your own question and tag it with `system.reactive`. — Stephen Cleary, Apr 01 '15 at 19:09
This solution does not preserve the ordering of results (just in case a reader is trying to achieve that). To preserve ordering look at the answers below. — bboyle1234, Aug 23 '19 at 09:32

score 0 · Answer 4 · answered Apr 12 '22 at 17:00

Building on reijerh's answer, I created an extension method.

public static IDisposable SubscribeAsync<TResult>(this IObservable<TResult> source, Func<Task> action) =>
            source.Select(_ => Observable.FromAsync(action))
                .Concat()
                .Subscribe();

If I understand this correctly, it should block until the async task finishes. But it will allow you to call SubscribeAsync and pass in your task. In my opinion this makes it a little more readable.

WhenSomethingHappened
    .SubscribeAsync(async () => { 
        await DoFirstAsyncThing(); 
        await DoSecondAsyncThing(); 
    });

How to call back async function from Rx Subscribe?

4 Answers4

Linked