在可观察到的取消上执行函数答案

【问题标题】：Execute function on observable cancellation在可观察到的取消上执行函数
【发布时间】：2019-05-04 14:40:51
【问题描述】：

我希望有一个 observable，当它取消订阅时它会调用一个函数，但只有当它被取消订阅时没有错误并且没有完成。我正在尝试构建的 observable 通常会与另一个 observable 竞争。我想要当另一个可观察的“获胜”时，这个执行一个函数。

我尝试了 finalize 运算符，但它总是执行。

播放.ts

import { timer } from "rxjs";
import { takeUntil, finalize } from "rxjs/operators";
import errorobs$ from "./errorobs";

export default function() {
  return timer(10000).pipe(
    takeUntil(errorobs$),
    finalize(finalFunc)
  );
}

function finalFunc() {
  console.log("final function executed");
}

errorobs.ts

import { fromEvent } from "rxjs";
import { map } from "rxjs/operators";

export default fromEvent(document.getElementById("errorBtn"), "click").pipe(
  map(() => {
    throw new Error("my error");
  })
);

我这里做了一个小demo https://codesandbox.io/s/q7pwowm4l6

单击开始以启动“可观察”。

点击取消让其他可观察者获胜

点击错误生成错误

【问题讨论】：

我希望它只在取消时调用函数。

标签： javascript rxjs rxjs6

【解决方案1】：

实现此目的的一种方法是使用自定义运算符，例如下面的onCancel()：

const {Observable} = rxjs

function onCancel(f) {
  return observable => new Observable(observer => {
    let completed = false
    let errored = false
    const subscription = observable.subscribe({
      next: v => observer.next(v),
      error: e => {
        errored = true
        observer.error(e)
      },
      complete: () => {
        completed = true
        observer.complete()
      }
    })
    return () => {
      subscription.unsubscribe()
      if (!completed && !errored) f()
    }
  })
}

// Test:
const {interval} = rxjs
const {take} = rxjs.operators

// This one gets cancelled:
const s = interval(200).pipe(
  onCancel(() => console.warn('s cancelled!'))
).subscribe(() => {})
setTimeout(() => s.unsubscribe(), 500) 

// This one completes before unsubscribe():
const q = interval(200).pipe(
  take(2),
  onCancel(() => console.warn('q cancelled!'))
).subscribe(() => {})
setTimeout(() => q.unsubscribe(), 500)

&lt;script src="//unpkg.com/rxjs@6/bundles/rxjs.umd.min.js"&gt;&lt;/script&gt;

【讨论】：

【解决方案2】：

正如您所描述的那样，它确实有效。 finalize 在链被处置时执行，即所有订阅者取消订阅、链出错或完成时。

RxJS Github 页面上已经存在关于此功能的问题：https://github.com/ReactiveX/rxjs/issues/2823

在上面的链接中，您可以看到一个自定义运算符的示例，它为finalize 运算符添加了理由。

我必须自己处理这个用例，并将这个运算符添加到我自己的 RxJS 运算符集合中：https://github.com/martinsik/rxjs-extra/blob/master/doc/finalizeWithReason.md

【讨论】：

我也喜欢你的回答。还有你的回购！