What's the difference between running code in Observable.Select() and Observable.Subscribe()?

ca9163d9 0
Viewed 57

When execute some Async function, we have to use Observable.Select(). For example,

var timer = Observable.Interval(TimeSpan.FromMilliseconds(100));
timer
    .Select(_ => Observable.FromAsync(() => DoSomething(cancellationToken)))
    .Concat()
    .Subscribe();

Using async function in Subscribe() will have problems.

timer.Subscribe(async tick => await DoSomething(cancellationToken));

Can the subscription logic be run in Select() always? (use .Select(...my logic...).Subscribe(); instead of .Subscribe(...my logic...);)

Why Subscribe()

system.reactive rx.net c#
1 Answers

There's a big semantic difference in the two blocks of code.

Try this complete program:

void Main()
{
    var cancellationTokenSource = new CancellationTokenSource();
    var cancellationToken = cancellationTokenSource.Token;
    
    var timer = Observable.Interval(TimeSpan.FromMilliseconds(100.0)).Take(5);

    // A
    timer
        .Select(_ => Observable.FromAsync(() => DoSomething(cancellationToken)))
        .Concat()
        .Subscribe();

    // B
    //timer.Subscribe(async tick => await DoSomething(cancellationToken));
}

public async Task<Unit> DoSomething(CancellationToken ct)
{
    Console.WriteLine("DoSomething Start");
    await Task.Delay(TimeSpan.FromMilliseconds(200.0));
    Console.WriteLine("DoSomething End");
    return Unit.Default;
}

When I run that I get this output:

DoSomething Start
DoSomething End
DoSomething Start
DoSomething End
DoSomething Start
DoSomething End
DoSomething Start
DoSomething End
DoSomething Start
DoSomething End

But when I comment out A and uncomment B I get this:

DoSomething Start
DoSomething Start
DoSomething Start
DoSomething End
DoSomething End
DoSomething Start
DoSomething Start
DoSomething End
DoSomething End
DoSomething End

.Subscribe does not understand what a async``/await` delegate is. It starts running and returns to the caller allowing the next value to be produced from the observable.

Using the Select/FromAsync/Concat method you ensure that the DoSomething is run one at a time, without overlapping.

You'd also get the correct results if you changed B to timer.Subscribe(tick => DoSomething(cancellationToken).Wait());. But this starts to show that there is something wrong with running async in a subscribe.

If you want to run async blocks in your subscribe then you need to give the compiler an extension method that does know how to await properly. Try this:

public static class ObservableEx2
{
    public static IDisposable Subscribe<T>(this IObservable<T> source, Func<T, Task> onNext) =>
        source
            .Select(t => Observable.FromAsync(() => onNext(t)))
            .Concat()
            .Subscribe();           
}

Now the code runs as expected when I run B.

Enigmativity0
Related