Execute function on observable cancellation

Question

I want to have an observable that when unsubscribed it calls a function but only when it is unsubscribed without error and without getting to complete. The observable I am trying to build usually gets raced with another observable. I want when the other observable "wins" this one executes a function.

I tried finalize operator but it executes always.

playback.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");
  })
);

I have made a small demo here https://codesandbox.io/s/q7pwowm4l6

click start to start "the observable".

click cancel to make the other observable win

click error to generate an error

Answer 1

One way to achieve this is using a custom operator, like my onCancel() below:

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>

Answer 2

It really works as you describe it. finalize is executed when the chain is being disposed which is when all subscribers unsubscribe, when the chain errors or when it completes.

There's already an issue on RxJS Github page for this feature: https://github.com/ReactiveX/rxjs/issues/2823

In the link above you can see an example of a custom operator that adds reason to the finalize operator.

I had to deal with this use-case myself and added this operator to my own collection of RxJS operators: https://github.com/martinsik/rxjs-extra/blob/master/doc/finalizeWithReason.md