Making an IObservable<T> that uses async/await return completed tasks in original order

Question

Suppose you have a list of 100 urls and you want to download them, parse the response and push the results through an IObservable:

public IObservable<ImageSource> GetImages(IEnumerable<string> urls)
{
    return urls
        .ToObservable()
        .Select(async url =>
        {
            var bytes = await this.DownloadImage(url);
            var image = await this.ParseImage(bytes);
            return image;
        });
}

I have some problems with this.

One is that it's bad etiquette to hammer a server with 100 requests at the same time -- ideally you would rate limit to maybe 6 requests at a given moment. However, if I add a Buffer call, due to the async lambda in Select, everything still fires at the same time.

Moreover, the results will come back in a different order than the input sequence of URLs, which is bad, because the images are part of an animation that will be displayed on the UI.

I've tried all kinds of things, and I have a solution that's working, but it feels convoluted:

public IObservable<ImageSource> GetImages(IEnumerable<string> urls)
{
    var semaphore = new SemaphoreSlim(6);

    return Observable.Create<ImageSource>(async observable =>
    {
        var tasks = urls
            .Select(async url =>
            {
                await semaphore.WaitAsync();
                var bytes = await this.DownloadImage(url);
                var image = await this.ParseImage(url);
            })
            .ToList();

        foreach (var task in tasks)
        {
            observable.OnNext(await task);
        }

        observable.OnCompleted();
    });
}

It works, but now I'm doing Observable.Create instead of just IObservable.Select, and I have to mess with the semaphore. Also, other animations that run on the UI stop when this is running (they're basically just DispatcherTimer instances), so I think I must be doing something wrong.

Answer 1

Give this a try:

urls.ToObservable()
    .Select(url => Observable.FromAsync(async () => {
        var bytes = await this.DownloadImage(url);
        var image = await this.ParseImage(bytes);
        return image;        
    }))
    .Merge(6 /*at a time*/);

What are we doing here?

For each URL, we're creating a Cold Observable (i.e. one that won't do anything at all, until somebody calls Subscribe). FromAsync returns an Observable that, when you Subscribe to it, runs the async block you gave it. So, we're Selecting the URL into an object that will do the work for us, but only if we ask it later.

Then, our result is an IObservable<IObservable<Image>> - a stream of Future results. We want to flatten that stream, into just a stream of results, so we use Merge(int). The merge operator will subscribe to n items at a time, and as they come back, we'll subscribe to more. Even if url list is very large, the items that Merge are buffering are only a URL and a Func object (i.e. the description of what to do), so relatively small.