Подождите, пока наблюдаемое завершится в пределах наблюдаемого

Question

У меня есть следующая структура данных:

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

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

Я использую Observable<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, но я все еще не мог найти работающее решение. Любая помощь будет оценена!

(код написан на угловом 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

Это хороший вопрос, и для меня это «большой шаг» к реальному пониманию всей силы наблюдаемых: наблюдаемых более высокого порядка или наблюдаемых, которые возвращают наблюдаемые.

Ваша ситуация идеально подходит для оператора 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 отобразит выбросы внешней Наблюдаемой (getRepos) во внутреннюю Наблюдаемую (getRepoContributors) и вернет новую Наблюдаемую, которая испускается, только когда завершается внутренняя Наблюдаемая. Другими словами, он сглаживает значения, которые передаются из одной наблюдаемой в другую, в один сжатый поток данных с возможностью подписки.

Обсерватории высшего порядка может быть трудно обернуть вашу голову, но в этом-то и заключается истинная сила Наблюдаемых. Я настоятельно рекомендую проверить некоторые другие операторы, такие как switchMap и concatMap на сайте, на который я ссылаюсь. Наблюдаемые могут стать сумасшедшими, когда используются в полной мере.

EDIT

Я неправильно прочитал исходный код и подумал, 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 в потоке, ожидая, что значение изменится в последовательных порядках с оператором.