ожидание наблюдаемой подписки внутри foreach с forkJoin

Question

Я пытаюсь заполнить массив в моем компоненте, называемом соглашениями, который является массивом соглашений.

У каждой организации есть список договоров, и у каждого договора есть идентификатор соглашения, с этим идентификатором я получил соглашение.

Я использую getOrganizationForUser, чтобы получить текущую организацию, а затем получить список контракта.,Затем я использую идентификатор соглашения из контракта для вызова второго API для получения соглашения.

В настоящее время мой код выглядит примерно так:

public getOrganizationForUser(): Observable<Organization> {
        return this.httpClient
            .get<Organization>(`${c.serviceBaseUrl.sp}/organizationByUser`)
            .pipe(catchError((err, source) => this.responseHandler.onCatch(err, source)));
    }

 public getById(id: number) {
        return this.httpClient
            .get<Convention>(`${c.serviceBaseUrl.sp}/conventions/` + id)
            .pipe(catchError((err, source) => this.responseHandler.onCatch(err, source)));
    }

ngOnInit() {
    this.OrganizationService.getOrganizationForUser().subscribe((organization: Organization) => {
        organization.contracts.forEach((contract) => {
            this.conventionService.getById(contract.conventionId).subscribe((convention: Convention) => {
                this.conventions.push(convention);
            })
        })
   })
}

Я понимаю, что могу создать массивиз наблюдаемых, и используйте Observable.forkJoin() для ожидания завершения всех этих асинхронных вызовов, но я хочу иметь возможность определить функцию обратного вызова подписки для каждого из вызовов, так как мне нужна ссылка на процесс.Любые идеи о том, как я могу подойти к решению этой проблемы?

Я пытался с этой функцией, но всегда возвращаюсь понять

getTasksForEachProcess(): Observable<Array<any>> {
    let tasksObservables = this.organizationService.getOrganizationForUser().pipe(map((organization: Organization) => {
        organization.contractOrganizations.map(contract => {
            return this.conventionService.getById(contract.conventionId).subscribe(convention =>
                this.conventions.push(convention)
            )
        });
    })
    );
    return forkJoin(tasksObservables);
};

ngOnInit() {
    this.getTasksForEachProcess().subscribe(item => {
        console.log(item);
    }
}

Answer 1

Я согласен с логикой Пиччи в обдумывании его ответа.Вот небольшое изменение к тому, что он предложил, хотя, как и он, я не проверил это строго, и могут быть некоторые ошибки.

Логика заключается в том, что в конечном итоге вы хотите получить массив соглашений имассив из наблюдаемых - это то, что делает zip.Итак, вот поток:

• получить организацию, затем создать поток наблюдаемых из массива organization.contracts, используя rxjs 'from.
• каждый элемент в этом потокебудет контрактом, который затем будет преобразован (с использованием map) в соглашение, основанное на поиске API, с использованием свойства contract.conventionId.
• весь этот результирующий поток наблюдаемых соглашений будет, наконец, преобразован обратно вмассив с помощью zip-пакета, и доставляется в виде наблюдаемой, на которую можно подписаться, в результате чего получается требуемый массив соглашений.

Вот код:

ngOnInit() {
    zip( this.OrganizationService.getOrganizationForUser()
        .pipe(
            map((organization: Organization) => 
                from(organization.contracts).pipe(
                    map(contract => this.conventionService.getById(contract.conventionId))
                )
            )
        )
    )
    .subscribe((conventions: Convention[]) => this.conventions = conventions)
}

Answer 2

Прежде всего, я не уверен, чего вы на самом деле пытаетесь достичь, так как я не понимаю, что вы подразумеваете под

Я хочу иметь возможность определять функцию обратного вызова подписки для каждогозвонков, так как мне нужна ссылка на процесс

В любом случае, в ситуации, подобной той, которую вы описываете, я бы сделал что-то вроде этого

public getOrganizationForUser(): Observable<Organization> {
        return this.httpClient
            .get<Organization>(`${c.serviceBaseUrl.sp}/organizationByUser`)
            .pipe(catchError((err, source) => this.responseHandler.onCatch(err, source)));
    }

 public getById(id: number) {
        return this.httpClient
            .get<Convention>(`${c.serviceBaseUrl.sp}/conventions/` + id)
            .pipe(catchError((err, source) => this.responseHandler.onCatch(err, source)));
    }

ngOnInit() {
    const obsOfObservables = this.OrganizationService.getOrganizationForUser()
    .pipe(
       map(organization => organization.contracts),
       map(contracts => contracts.map(contract => this.conventionService.getById(contract.conventionId)))
    );

    obsOfObservables
    .pipe(
       switchMap(conventionObservables => forkJoin(conventionObservables))
    )
    .subscribe(
       conventions => { // do stuff with conventions }
    )
}

Ключевые моменты здесьследующие:

Через getOrganizationForUser() вы получаете Наблюдаемое, которое испускает Организацию.Первое, что вы делаете, вы превращаете объект, испускаемый наблюдаемой, в Массив контрактов с первым оператором map.

Второй оператор map преобразует Массив контрактов в Массив наблюдаемых соглашений,Для выполнения этого преобразования мы используем метод Array map в операторе map объекта Observable.Это может немного сбивать с толку, но это стоит понять.

Если мы остановимся здесь, то у нас будет obsOfObservables, то есть Наблюдаемая, которая испускает Массив Наблюдаемых.

Мы тогдапередать массив наблюдаемых, выданных obsOfObservables, в функцию forkJoin, которая сама по себе возвращает наблюдаемую.Так как мы на самом деле заинтересованы в том, что уведомляет Observable, возвращаемое forkJoin, то есть нас интересуют соглашения, то нам нужно переключить с первой Observable на вторую, и это делается с помощьюswitchMap operator.

Чистый результат - Observable, который возвращает массив соглашений.Учтите, что константа obsOfObservables была добавлена ​​как попытка прояснить причину, и она совершенно не нужна (как сказал бы Барни Панофски).

Я не симулировал все это,так что я надеюсь, что я не вставил ошибки, но более или менее это мыслительный процесс, который я бы использовал в этом случае.

Последнее примечание, обычно будьте подозрительны, если у вас есть subscribe в пределах subscibe.