使用响应式扩展(Reactive Extensions, Rx)编排的异步REST API调用序列

Question

我正在尝试构建一个服务层，它通过使用响应式扩展进行REST API调用的协调序列来提供一些数据。为了这个问题，我正在使用可观察的Octokit.net库试图从GitHub的REST API中提取一些数据(请不要选择使用Octokit库的方式)。具体地说，我想在这个场景中检索用户的详细信息、存储库列表和ssh密钥列表。因此，我想协调以下操作顺序:

。通过调用用户API获取用户详细信息B.结果出来后，带上用户。登录并并行启动两个REST调用，以检索repos列表和ssh密钥列表。c.当api调用repos和ssh密钥完成时，在结果序列中产生一个项目(GithubUserDTO)，该项目包含从a)和b)的api结果中填充的字段

下面是我到目前为止生成的代码，它似乎由于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;
                });
            });
    }

GithubUserDto的样子:

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)当API调用Repository和SshKey独立并以异步模式完成时，我如何确保在GithubUserDto序列中产生一个项目?Zip似乎不是这个场景中的选项。

2)我怎么能继续生产项目GithubUserDto的结果序列，即使调用一个"次要"API(如SshKey)异步失败(例如由于网络问题)?

Answer 1

看起来SshKeys和Repositories一次返回一个，但您真正需要的是一个列表。通过在Observable上使用ToList，你可以从IObservable<T>得到IObservable<IList<T>>。

但在此之前，您需要处理错误情况。如果只是想忽略异常，可以使用Catch操作符的重载，该操作符接受Observable。你可以将Empty可观察对象传递给该操作符，当发生异常时，它将忽略它并切换到空的可观察对象，而不是原始源。

 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
                    };
                }
            );
        });
}