как получить доступ к предыдущим значениям mergeMap из rxjs

Question

Я учусь использовать RXJS.В этом сценарии я создаю несколько асинхронных запросов с использованием rxjs.На последнем mergeMap я хотел бы получить доступ к параметрам первого mergeMap.Я исследовал этот параметр, используя Global или withLatest, но ни один из этих вариантов здесь не подходит.

const arraySrc$ = from(gauges).pipe(
    mergeMap(gauge => {
      return readCSVFile(gauge.id);
    }),
    mergeMap((csvStr: any) => readStringToArray(csvStr.data)),
    map((array: string[][]) => transposeArray(array)),
    mergeMap((array: number[][]) => forkJoin(uploadToDB(array, gauge.id))),
    catchError(error => of(`Bad Promise: ${error}`))
  );

readCSVFile - это асинхронный запрос, который возвращает наблюдаемый для чтения CSV изУдаленный сервер.

readStringToArray - это еще один асинхронный запрос, который возвращает наблюдаемую информацию для преобразования string в Arrays

transposeArray, просто выполняет транспонирование

uploadToDB это асинхронный запрос к БД, который требует gague.id из первой таблицы слияния.

Как мне это получить?Было бы здорово посоветовать, почему то, как я это делаю, плохо.

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

const arraySrc$ = from(gauges).pipe(
    mergeMap(gauge => readCSVFile(gauge.id)),
    mergeMap(({ data, gaugeId }: any) => readStringToArray(data, gaugeId)),
    map(({ data, gaugeId }) => transposeArray(data, gaugeId)),
    mergeMap(({ data, gaugeId }) => uploadToDB(data, gaugeId)),
    catchError(error => of(`Bad Promise: ${error}`))
  );

Answer 1

Почему бы вам не сделать это просто?

const arraySrc$ = from(gauges).pipe(
    mergeMap(gauge => readCSVFile(gauge.id).pipe(
         mergeMap((csvStr: any) => readStringToArray(csvStr.data)),
         map((array: string[][]) => transposeArray(array)),
         mergeMap((array: number[][]) => forkJoin(uploadToDB(array, gauge.id)))
    )),
    catchError(error => of(`Bad Promise: ${error}`))
);

Вы также можете обернуть внутреннюю наблюдаемую в функцию:

uploadCSVFilesFromGaugeID(gaugeID): Observable<void> {
     return readCSVFile(gaugeID).pipe(
         mergeMap((csvStr: any) => readStringToArray(csvStr.data)),
         map((array: string[][]) => transposeArray(array)),
         mergeMap((array: number[][]) => forkJoin(uploadToDB(array, gaugeID))
     );
}

Чтобы сделать это в конце:

const arraySrc$ = from(gauges).pipe(
    mergeMap(gauge => uploadCSVFileFromGaugeID(gauge.id)),
    catchError(error => of(`Bad Promise: ${error}`))
);