如何在链式可观察量之间传递结果

抽象问题：每次源 Observable 发出事件时，都需要触发一系列 API 调用和 Angular 服务。其中一些调用取决于之前的结果。

在我的示例中，源 ObservablestartUpload$触发一系列依赖调用。

使用解构可以这样写：

this.startUploadEvent$.pipe(
      concatMap(event => this.getAuthenticationHeaders(event)),
      map(({ event, headers }) => this.generateUploadId(event, headers)),
      tap(({ event, headers, id }) => this.emitUploadStartEvent(id, event)),
      concatMap(({ event, headers, id }) => this.createPdfDocument(event, headers, id)),
      concatMap(({ event, headers, id, pdfId }) => this.uploadBilderForPdf(event, pdfId, headers, id)),
      mergeMap(({ event, headers, id, pdfId, cloudId }) => this.closePdf(cloudId, event, headers, id, pdfId)),
      tap(({ event, headers, id, pdfId, cloudId }) => this.emitUploadDoneEvent(id, event, cloudId)),
).subscribe()

它读起来几乎像是一种命令式的方法。但它也存在一定的问题：

• 解构链在代码中不断重复，并且变得越来越长{ event, headers, id, pdfId, cloudId }
• 方法（如generateUploadId(event, headers)）需要接收所有先前的值，以便它们能够将它们传递到下一个管道，即使方法本身不需要它
• 内部可观察对象（在方法内）需要映射值，以便进一步的管道阶段可以破坏它们：

_

private closePdf(cloudId, event, headers, id, pdfId) {
    return this.httpClient.post(..., { headers } )
        .pipe(
             //...,
             map(() => ({ event, headers, id, pdfId, cloudId }))
        )
}

如果编译器能够处理样板文件（就像async await）编写如下代码（没有上述问题）：

private startUpload(event: StartUploadEvent) {
    const headers = this.getAuthenticationHeaders(event)
    const id = this.generateUploadId()

    this.emitUploadStartEvent(id, event)

    const pdfId = this.createPdfDocument(event, headers, id)
    this.uploadBilderForPdf(event, pdfId, headers, id)

    const cloudId = this.closePdf(headers, pdfId)
    this.emitUploadDoneEvent(id, event, cloudId)

    return cloudId
  }

如何在链式可观察量之间传递结果而不出现我提到的问题？我错过了 rxjs 概念吗？

您当然不应该让您的方法接受与它们无关的参数！

对于你的主要问题：

如何在链式可观察量之间传递结果而不出现我提到的问题？

使用单一范围（嵌套管道）

下面的代码相当于您的示例代码，无需传递不必要的属性。先前返回的值可以通过链下游的函数调用来访问：

1   startUploadEvent$.pipe(
2     concatMap(event => getAuthenticationHeaders(event).pipe(
3       map(headers => generateUploadId(event, headers).pipe(
4         tap(id => emitUploadStartEvent(id, event)),
5         concatMap(id => createPdfDocument(event, headers, id)),
6         concatMap(pdfId => uploadBilderForPdf(event, pdfId)),
7         tap(cloudId => closePdf(cloudId, event))
8       ))
9     ))
10  ).subscribe();

注意如何event and headers可以到达下游。它们不需要传递到不需要它们的函数中。

我错过了 rxjs 概念吗？

或许。？并不真地... ：-）

诀窍是附加一个.pipe有效地对操作员进行分组，以便他们都可以访问输入参数。

通常，我们尝试将代码保持在平面内.pipe:

1   const greeting$ = userId$.pipe(
2     switchMap(id => http.get(`/users/${id}`)),
3     map(response => response.data.userName),
4     map(name => `Hello ${name}!`),
5     tap(greeting => console.log(greeting))
6   );

但该代码实际上与以下代码没有什么不同：

1   const greeting$ = userId$.pipe(
2     switchMap(id => http.get(`/users/${id}`).pipe(
3       map(response => response.data.userName),
4       map(name => `Hello ${name}! (aka User #${id})`)
5     )),
6     tap(greeting => console.log(greeting))
7   );

但是，在第二种情况下，第 4 行可以访问name和id，而在第一种情况下它只能访问name.

注意第一个的签名是userId$.pipe(switchMap(), map(), map(), tap())

第二个是：userId$.pipe(switchMap(), tap()).