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

Question

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

我尝试了 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");
}

错误对象.ts

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

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

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

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

单击“取消”以使其他可观察的获胜

点击error生成错误

Answer 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)

<script src="//unpkg.com/rxjs@6/bundles/rxjs.umd.min.js"></script>