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

Question

我想有一个可观察的，当取消订阅时它调用一个函数，但只有当它取消订阅时没有错误并且没有完成。我尝试构建的 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");
  })
);

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

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

点击取消让其他observable获胜

点击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>