Why C# Rx Subscribe() function doesn't work with "async" keyword?

Question

I've got this code snippet:

static void Main(string[] args)
{
    Observable.Range(1, 5).Subscribe(async x => await DoTheThing(x));
    Console.WriteLine("done");
}

static async Task DoTheThing(int x)
{
    await Task.Delay(TimeSpan.FromSeconds(x));
    Console.WriteLine(x);
}

I hope it will loop 5 times and after each loop there'll be a line printed as

1
2
3
4
5

But supprisingly, this will print "done" and terminate at once. Seems that async+await didn't wait the Task.Delay and quit.

Semantics doesn't seem to have problem, so where did I get wrong about Subscribe or async, how to fix it to fulfill my request of calling asynchrous tasks from Rx?

Thanks.

Answer 1

It's not blocking, because it's well - asynchronous. Your code generates five tasks, all running parallel, all completing at various times.

But they do not block the Main function. If you just add a Console.ReadKey() as your last line, you'll see that your code does run in the background. It prints.

static void Main(string[] args)
{
    Observable.Range(1, 5).Subscribe(async x => await DoTheThing(x));
    Console.WriteLine("done");
    Console.ReadKey();
}

But suppose you want to wait until all of them were done. What then?

Of course, there's .Wait(), but that's blocking. Let's observe all of our tasks as observables.

We'll use C# 7's async Main while we're at it.

static async Task Main(string[] args)
{
    await Observable.Range(1, 5)
        .Select(x => DoTheThing(x).ToObservable())
        .Merge();                

    Console.WriteLine("done");
}

This works exactly like you'd expect it to.