Nested Observables: что я делаю не так?

Question

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

Ниже приведен мой код, которыйв настоящее время не работает.Я не уверен, что я делаю неправильно.Что я должен изменить, чтобы этот код работал?

    this.esService.search(JSON.stringify(query)).subscribe(response => {
        let temp;
        this.esResponse = response;
        let result = this.esResponse.hits.hits;
        return from(result).pipe(
            concatMap(item => 
            this.productService.getProductById(this.schemaId, item['_source']['consensusAssessmentId']).pipe(
                map(resp => {
                const answers = resp['answers'];
                const meets_count = answers.reduce((acc, cur) => cur.value === "Meets" ? ++acc : acc, 0);
                const needs_count = answers.reduce((acc, cur) => cur.value === "Needs Improvement" ? ++acc : acc, 0);
                item['_source']['meets'] = meets_count.toString();
                item['_source']['needs'] = needs_count.toString();
            })
        ))

        ).subscribe(resp => {
            temp = resp; 
            this.rows = temp.map(
                this.esService.convertToPE
            ); 
        })    


    })

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

    this.esService.search(JSON.stringify(query)).subscribe(response => {
        let temp = [];
        this.esResponse = response;
        let result = this.esResponse.hits.hits;
        result.forEach(item => {
            this.productService.getProductById(this.schemaId, item['_source']['consensusAssessmentId']).subscribe(resp => {
                const answers = resp['answers'];
                const meets_count = answers.reduce((acc, cur) => cur.value === "Meets" ? ++acc : acc, 0);
                const needs_count = answers.reduce((acc, cur) => cur.value === "Needs Improvement" ? ++acc : acc, 0);
                item['_source']['meets'] = meets_count.toString();
                item['_source']['needs'] = needs_count.toString();
            })
           temp.push(item);
        })          
        this.rows = temp.map(
            this.esService.convertToPE
        );

    });

Answer 1

Похоже, у вас есть представление о том, как работают наблюдаемые, но давайте внесем некоторые изменения в подход к вашей проблеме, упростив ваш подход.Во-первых, вместо вложения / объединения в цепочку нескольких subscribe() методов мы используем конвейерных операторов .

Затем мы используем оператор RxJS mergeMap для отображения наблюдаемых значений из search() во внутреннюю наблюдаемую, которая будет потребляться getProductById().

Внутри блока mergeMap() нам потребуется выполнить итерацию по массиву, представленному константной переменной result.Мы перебираем все элементы в массиве result, затем помещаем каждый наблюдаемый элемент из getProductById() в observablesList.Затем мы используем оператор forkJoin () , который ожидает завершения итераций перед возвратом всех наблюдаемых.

Не забудьте импортировать необходимые операторы в ваш класс / компонент.

import { forkJoin } from 'rxjs';
import { mergeMap } from 'rxjs/operators';

this.esService.search(JSON.stringify(query)).pipe(
    mergeMap(response => {
      this.esResponse = response;
      const result = this.esResponse.hits.hits;
      const observablesList = [];
      observablesList = result.map(obj => {
        const id = obj['_source']['consensusAssessmentId'];
        return this.productService.getProductById(this.schemaId, id);
      });
      return forkJoin(observablesList);
    })
  ).subscribe(resp => {
    console.log(resp);
     // handle the rest here
     const answers = resp['answers'];
     const meets_count = answers.reduce((acc, cur) => cur.value === "Meets" ? ++acc : acc, 0);
     const needs_count = answers.reduce((acc, cur) => cur.value === "Needs Improvement" ? ++acc : acc, 0);
     item['_source']['meets'] = meets_count.toString();
     item['_source']['needs'] = needs_count.toString();

     // more things
  })
});

Answer 2

Я не уверен, что у меня есть то, что ты хочешь сделать, но я думаю, ты хочешь что-то вроде этого:

this.esService.search(JSON.stringify(query)).pipe(
  tap((response) => this.esResponse = response),
  switchMap((response) => {
    const arrayOfObservables: any[] = response.hits.hits.map((item) =>
      this.productService.getProductById(this.schemaId,
       item['_source']['consensusAssessmentId']).pipe(tap((resp) => {
          const answers = resp['answers'];
          const meets_count = answers.reduce((acc, cur) => cur.value === "Meets" ? ++acc : acc, 0);
          const needs_count = answers.reduce((acc, cur) => cur.value === "Needs Improvement" ? ++acc : acc, 0);
          item['_source']['meets'] = meets_count.toString();
          item['_source']['needs'] = needs_count.toString();
        }))
    );
    return forkJoin(arrayOfObservables);

  })).subscribe((resp) => this.rows = resp.map(
    this.esService.convertToPE));