如何知道 switchMap 是否取消了内部 Observable 的订阅？

Question

由于switchMap 可能会在有新项目进入时切断（取消订阅）内部 Observable，因此我们预计会出现以下情况：

  const items = from([1, 2])
  const seen = []
  const derived = items.pipe(
    switchMap(item =>
      concat(
        of(`now: ${item}`),
        of(`end: ${item}`).pipe(delay(10))
      )
    )
  )

  derived.subscribe(next => seen.push(next))

导致seen 拥有：

['now: 1', 'now: 2', 'end: 2']
           ^ unsubscribed to 1 here, as planned!

问题是，有没有办法在第 1 项被取消时收到通知？原因是，如果取消太频繁，我想知道。

我已经尝试过修改内部 Observables 的方法，但基本上我尝试在它们“最后”做的任何事情在取消订阅时都不会发生。观察者规范有 next、completed 和 error，但没有“提前结束”，所以我不知道该尝试什么。

我希望能够向 switchMap 传递一个附加函数，当内部可观察对象被取消并提供参数时，该函数将由操作员调用。

  const derived = items.pipe(
    switchMap(item => concat(
      of(`now: ${item}`),
      of(`end: ${item}`).pipe(delay(10))
    ),
    (item) => console.error(`${item} was cut off.`))
  )

Answer 1

编辑：RxJS GitHub 页面上有一个问题，讨论了 finalize 运算符的此功能。 https://github.com/ReactiveX/rxjs/issues/2823

由于switchMap 取消订阅内部的 Observable，它不会收到任何完整的通知。

相反，您可以添加一个处置处理程序，或者简单地说使用 finalize() 运算符为您执行此操作，并在处置链时调用它。这适用于完成和取消订阅事件。

switchMap(item =>
  concat(
    of(`now: ${item}`),
    of(`end: ${item}`).pipe(delay(10))
  ).pipe(
    finalize(() => /* switchMap unsubscribed or concat completed */),
  )
)

请注意，当您取消订阅时，这意味着您不想再从 Observable 接收任何项目。这意味着您在退订时无法从退订的 Observable 收到任何新项目。例如，defaultIfEmpty 运算符在源完成时发出一个项目，而不发出任何适用于完整通知但不能工作的东西，你只是取消订阅。