等待 Observable 在 Observable 内完成

Question

我有以下数据结构：

export class Repo {
    id: number;
    name: string;
    contributors: Contributor[];
}

export class Contributor {
    id: number;
    login: string;
}

我正在使用Observable<Repo> 获取所有回购数据，但我想在外部Repo 被认为完全发出之前在内部调用另一个可观察的Observable<Contributor> 来填充所有贡献者。我不知道该怎么做。我有以下代码。

private repos: Repo[] = [];

getRepos(orgName: string): void {
    const repoBuild: Repo[] = [];
    this.githubService.getRepos(orgName).pipe(
        // this here won't wait for all contributors to be resolved
        map(repo => this.getRepoContributors(orgName, repo))
    ).subscribe(
        repo => repoBuild.push(repo),
        error => {
            this.repos = [];
            console.log(error);
        },
        () => this.repos = repoBuild
    );
}

// fetches contributors data for the repo
private getRepoContributors(orgName: string, repo: Repo): Repo {
    const contributors: Contributor[] = [];
    repo.contributors = contributors;
    this.githubService.getRepoContributors(orgName, repo.name)
        .subscribe(
            contributor => {
                // add to the total collection of contributors for this repo
                contributors.push(contributor);
            },
            error => console.log(error),
            () => repo.contributors = contributors
        );
    return repo;
}

我承认我对Observable 的理解是有限的，我为此苦苦挣扎了好几个小时。我尝试在 StackOverflow 上找到适合我的东西，但我仍然找不到有效的解决方案。任何帮助将不胜感激！

（代码是用Angular 5编写的）

解决方案：

我使用了下面的@joh04667 建议并最终让它发挥作用。这是我的做法：

getRepos(orgName: string): void {
    const repoBuild: Repo[] = [];
    this.githubService.getRepos(orgName).pipe(
        // mergeMap() replaces `repo` with the result of the observable from `getRepoContributors`
        mergeMap(repo => this.getRepoContributors(orgName, repo))
    ).subscribe(
        repo => repoBuild.push(repo),
        error => {
            this.repos = [];
            console.log(error);
        },
        () => this.repos = repoBuild
    );
}

// fetches contributors data for the repo
private getRepoContributors(orgName: string, repo: Repo): Observable<Repo> {
    repo.contributors = []; // make sure each repo has an empty array of contributors
    return this.githubService.getRepoContributors(orgName, repo.name).pipe(
        // tap() allows us to peek on each contributor and add them to the array of contributors
        tap(contributor => {
            // add to the total collection of contributors for this repo
            repo.contributors.push(contributor);
        }),
        // only picks the last contributor and replaces him/her with the repo
        last(
            () => false,
            () => repo,
            repo
        )
    );
}

在我使用last() 的最后一部分中，我基本上告诉Observable，即使它会处理所有值，我只会使用最后一个值。最后一个是Contributor 类型，但我将其替换为默认值（repo），它允许我将返回类型从Observable<Contributor> 更改为Observable<Repo>，这正是我需要的更高级别可观察。

Answer 1

这是一个很好的问题，对我来说，这是真正理解 Observables 的全部功能的“重要一步”：高阶 Observables，或返回 Observables 的 Observables。

您的情况非常适合mergeMap / flatMap 运算符：

getRepos(orgName: string): void {
    const repoBuild: Repo[] = [];
    this.githubService.getRepos(orgName).pipe(
        // this will map the emissions of getRepos to getRepoContributors and return a single flattened Observable
        mergeMap(repo => this.getRepoContributors(orgName, repo))
    ).subscribe(
        repo => repoBuild.push(repo),
        error => {
            this.repos = [];
            console.log(error);
        },
        () => this.repos = repoBuild
    );
}

mergeMap 会将外部 Observable (getRepos) 的发射映射到内部 Observable (getRepoContributors) 并返回一个仅在内部 Observable 完成时发射的新 Observable。换句话说，它将从一个 Observable 传递到另一个 Observable 的值扁平化为一个简洁的可订阅数据流。

高阶 Observable 可能难以理解，但这才是 Observable 真正强大的地方。我强烈建议在我链接的网站上查看其他一些运营商，例如 switchMap 和 concatMap。充分利用 Observables 会变得非常强大。

编辑

我误读了原始代码，并认为 getRepoContributors 正在返回一个 Observable。漫长的工作日让我很煎熬。我将在这里重构：

map 适合在与mergeMap 合并之前更改值。 getRepoContributors 的前几行可以在那里完成。由于mergeMap 需要一个 Observable 返回给它，我们可以简化一下：

private repos: Repo[] = [];

getRepos(orgName: string): void {
    const repoBuild: Repo[] = [];
    this.githubService.getRepos(orgName).pipe(
        map(repo => {
            repo.contributors = [];
            return repo;
          }),
        mergeMap(repo => this.githubService.getRepoContributors(orgName, repo.name)),
        map(contributor => {
            repo.contributors.push(contributor)
          })
    ).subscribe(
        repo => repoBuild.push(repo),
        error => {
            this.repos = [];
            console.log(error);
        },
        () => this.repos = repoBuild
    );
}

// don't need second function anymore

我们可以在流中map 期望值随着操作员按顺序更改。