Should we use Reactive Extension's Observable

Question

I have a general question regarding designing the architecture of my cross-platform solution (Xamarin.Forms and ASP.NET Core).

I have a requirement of calling multiple API endpoints (similar to a blog aggregation app) and I want to create a shared REST API wrapper library using Reactive Extension or Refit which will be consumed by both Xamarin.Forms and ASP.NET Core website. As per the Reactive Extension (Rx.NET) documentation - "An advantage of this approach (Observable) is that when you have a bunch of tasks that are not dependent on each other, you can start them all at the same time rather than waiting for each one to finish before starting the next one". But in my case, I want to start multiple API endpoints call in background but want to wait until all completes which means that the tasks are dependent here.

Does it make sense to use Rx.NET's Observable pattern here as my observer as a subscriber will not receive any notification from Observable until all tasks complete? What are the pros and cons here if I use Task Parallel library (with Task.WhenAll()) rather than Rx.NET's Observable considering I need to consume it on both mobile (X.F) and web (ASP.NET Core) platform.

Answer 1

Rx is awesome for calling multiple API end points simultaneously.

You can chose to process each result as it arrives or have the query bundle the results as an array and return them all at the end.

Let's start with an array of API calls, like this:

var api_calls = new Func<string>[]
{
    () => "Hello",
    () => "World",
};

It's a trivial array of functions that return strings.

Now the simplest Rx query looks like this:

IObservable<string> query1 =
    from f in api_calls.ToObservable()
    from r in Observable.Start(() => f())
    select r;

When you subscribe to this query you get the values one at a time in the order in which the calls return (not the order in which they were requested).

You can write this query like this instead:

IObservable<string> query1 =
    api_calls
        .ToObservable()
        .SelectMany(f =>
            Observable
                .Start(() => f()));

This way of writing the query allows us to do a few more things.

If I want return all of the results in at the end I just do this:

IObservable<string[]> query2 =
    api_calls.ToObservable()
        .SelectMany(f => Observable.Start(() => f()))
        .ToArray();

Now I get an observable that still queries simultaneously, but returns only a single string[] as its only value.

The problem now is that it can be out of order. To restore the order (should you wish to have the results match the original order of the Func<string>[] source you can do this:

IObservable<string[]> query =
    api_calls
        .ToObservable()
        .SelectMany((f, i) =>
            Observable
                .Start(() => (i, f())))
        .ToArray()
        .Select(xs =>
            xs
                .OrderBy(x => x.Item1)
                .Select(x => x.Item2)
                .ToArray());

Running that query I get an array with { "Hello", "World" }, in that order, each and every time I subscribe.