Orchestrated sequence of async REST API calls using Reactive Extensions (Rx)

Question

I am trying to build a service layer which provides some data by making a coordinated sequence of REST API calls using Reactive Extensions. For the sake of this question, I am using observable Octokit.net library trying to pull some data out of GitHub's REST API (don't pick on Octokit library way of using it please). Specifically, I want to retrieve a user's details, list of repositories and list of ssh keys, for this scenario. So I want to coordinate the following order of operations:

a. bring user details by calling the user api b. when the result comes, take user.Login and start in parallel two REST calls to retrieve list of repos and list of ssh keys. c. when both api calls for repos and ssh keys finish, produce an item (GithubUserDTO) in the result sequence which contains fields populated from the API results from a) and b)

The following is the code I produced so far, which seems to suffer due to Zip.

    public IObservable<GithubUserDto> Get(string username)
    {
        return githubObservableClient.User.Get(username)
            .SelectMany(user =>
            {
                var userRepos = githubObservableClient.Repository.GetAllForUser(user.Login /* for the sake of demo, I assume I don't have user.Login from beginning */);
                var sshKeys = githubObservableClient.SshKey.GetAll(user.Login  /* for the sake of demo, I assume I don't have user.Login from beginning */);

                return userRepos.Zip(sshKeys, (repo, sshkey) =>
                {
                    var userDto = new GithubUserDto() {Id = user.Id, Name = user.Name};
                    userDto.Repositories.Add(repo.FullName);
                    userDto.SshKeys.Add(sshkey.Key.Substring(0, Math.Min(20, sshkey.Key.Length)));
                    return userDto;
                });
            });
    }

Where GithubUserDto looks like:

public class GithubUserDto
    {
        public GithubUserDto()
        {
            Repositories = new List<string>();
            SshKeys = new List<string>();
        }

        public int Id { get; set; }
        public string Name { get; set; }
        public List<string> Repositories { get; set; }
        public List<string> SshKeys { get; set; }
    }

1) How can I make sure that an item in GithubUserDto sequence will be produced when API calls to Repository and SshKey will finish independently and in async mode? Zip does not seem to be the option in this scenario.

2) How can I continue producing items of GithubUserDto in the result sequence even if a call to one of the "secondary" API (like SshKey) fails async (e.g. due to network problems)?

Answer 1

It looks like the SshKeys and Repositories are being returned one at a time, but what you really need is a list. You can get an IObservable<IList<T>> from an IObservable<T> by using ToList on the Observable.

But before you do that, you'll want to handle the error cases. If you simply want to ignore exceptions, you can use the overload of the Catch operator which accepts an Observable. You can pass an Empty observable to that operator, and when an exception occurs, it will ignore it and switch to the empty observable instead of the original source.

 public IObservable<GithubUserDto> Get(string username)
{
    return githubObservableClient.User.Get(username)
        .SelectMany(user =>
        {
            var userRepos = githubObservableClient.Repository
                .GetAllForUser(user.Login)
                .Catch(Observable.Empty<Repository>()) // empty if error
                .ToList(); // Put all Repositories into a list

            var sshKeys = githubObservableClient.SshKey
                .GetAll(user.Login)
                .Catch(Observable.Empty<SshKey>()) // empty if error
                .ToList(); // Put all Repositories into a list

            return Observable.Zip(
                userRepos,
                sshKeys,
                (repos, keys) =>
                {
                    return new GithubUserDto() {
                        Id              = user.Id,
                        Name            = user.Name,
                        Repositories    = repos,
                        SshKeys         = keys
                    };
                }
            );
        });
}