RxJS：将对象键映射到可观察对象答案

【问题标题】：RxJS: Mapping object keys to observablesRxJS：将对象键映射到可观察对象
【发布时间】：2019-06-02 08:57:03
【问题描述】：

我有一个向用户显示问题的应用。问题的草稿从 SharePoint 列表加载。每个草稿都包含一个键，用于从另一个 SharePoint 列表加载对问题的正确响应。以下是我目前的实现方式：

interface QuestionDraft {
  title: string;
  responseKey: string;
}

interface Question {
  title: string;
  responses: string[];
}

const drafts: QuestionDraft[] = [];
const questions: Question[] = [];

// stub
private getDrafts(): Observable<QuestionDraft> {
    return from(drafts);
}

// stub
private getResponses(key: string): Observable<string> {
    return of(key, key, key);
}

main(): void {
    getDrafts().subscribe(
      data => {
        const res: string[] = [];
        getResponses(data.responseKey).subscribe(
          d => res.push(d),
          error => console.error(error),
          () => questions.push({
            title: data.title,
            responses: res
          })
        );
      }, error => console.error(error),
      () => console.log(questions)
    );
}

这个解决方案工作正常，但我认为main() 中的代码看起来很乱。有没有更简单的方法来做同样的事情，例如使用mergeMap 或类似的东西？

【问题讨论】：

标签： javascript typescript rxjs observable

【解决方案1】：

您可以使用 mergeMap 映射到一个新的 Observable 并使用 toArray 将发出的值收集到一个数组中。使用catchError 处理流中的错误并映射到另一个可观察到的错误。

此代码将与您的代码一样工作，其中包含所有问题的发出问题数组，直到 getDrafts 引发错误并排除 getResponses 引发错误的问题。

getDrafts().pipe(
    mergeMap(draft => getResponses(draft.responseKey).pipe(
        toArray(),
        map(responses => ({ title: draft.title, responses } as Question)),
        catchError(error => { console.error(error); return EMPTY; })
    )),
    catchError(error => { console.error(error); return EMPTY; }),
    toArray()
).subscribe(qs => { console.log(qs); questions = qs; })

请记住，最终数组中的questions 不一定与传入的drafts 的顺序相同。顺序取决于getResponses Observable 完成特定draft 的速度。（这与您当前的代码行为相同）

为确保questions 与drafts 的顺序相同，您可以使用concatMap 而不是mergeMap。但这可能会减慢任务的整体执行速度，因为只有在前一个 draft 的响应完成后才会获取下一个 draft 的响应。

【讨论】：

【解决方案2】：

您可以尝试使用flatMap 使其更清洁。 RxJs Observables nested subscriptions?

this.getDrafts()
    .flatMap(function(x){return functionReturningObservableOrPromise(x)})
    .flatMap(...ad infinitum)
    .subscribe(...final processing)

【讨论】：

【解决方案3】：

如果你使用 RxJS 版本 6，你必须使用 pipe() 方法并将 flatMap 转换为 mergeMap。

在 rxjs 6 中，@emcee22 的示例如下：

this.getDrafts()
 .pipe(
   .mergeMap(function(x){return functionReturningObservableOrPromise(x)}),
   .mergeMap(...ad infinitum)
 ).subscribe(...final processing)

【讨论】：