angular - 等待 Observable 在 Observable 内完成

Question

我有以下数据结构:

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

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

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

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>这正是我对更高级别的 Observable 所需要的。

Answer 1

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

您的情况非常适合 mergeMap / flatMap运算符(operator):

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 的值扁平化为一个简洁的可订阅数据流。

高阶 Observables 可能很难让您全神贯注，但这确实是 Observables 真正强大的地方。我强烈建议在我链接的网站上查看一些其他运算符，如 switchMap 和 concatMap。当 Observable 发挥到极致时，它会变得异常强大。

编辑

我误读了原始代码并认为 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

我们可以在流中映射，期望值随运算符按顺序改变。