Does an observable keep emitting values after observer has unsubscribed?

Question

Let's consider the following example code:

gude() {
  const digestor$ = new Observable(subscriber => {
    for (let i = 0; i < 4711; i++) {
      setTimeout(() => {
        const hash = createHashWithNLeadingZeroes(i);
        subscriber.next(hash);
      }, i);
    }
  });

  const subscription = digestor$.subscribe(
    _ => {
      if (subscription) {
        subscription.unsubscribe();
      }
    }
  );
}

Within the function gude() a new observable is created which emits hash values that have the first n leading values set to zero. An observer subscribes to that observable and immediately unsubscribes. Let's assume that the function createHashWithNLeadingZeroes() takes quite a bit of time to generate a response.

IMHO the following is happening here:

(1) A new Observable is created and the function describing the behavior of the Observable is internally stored in the property _subscribe (https://github.com/ReactiveX/rxjs/blob/master/src/internal/Observable.ts: 37-41).

(2) When subscribe() is called, first the Observer is wrapped in a Subscriber object, and second that Subscriber is applied to the _subscribe function which holds the Observable's logic. _subscribe() returns quickly as only 4711 timeout's are set and a Subscription object is returned (https://github.com/ReactiveX/rxjs/blob/master/src/internal/Observable.ts: 206-250).

The Subscriber basically intercepts calls to next(), error(), and complete() and only forwards to the actual observer when internally the property isStopped is not set (https://github.com/ReactiveX/rxjs/blob/master/src/internal/Subscriber.ts: 90-128).

(3) As soon as the variable subscription is set, unsubscribe() is called. Among others, this will lead to setting isStopped to true so that the Subscriber will not forward the hashes to the observer anymore (https://github.com/ReactiveX/rxjs/blob/master/src/internal/Subscriber.ts: 130-136).

According to that logic the Observable however would still continue doing its job until all 4711 hashes are computed for no purpose as the observer's methods are turned to noops. Ultimately, such a behavior can affect the application's performance depending on the amount of subscriptions and workload of the Observable's. I find it somewhat hard to believe that the described is correct. Which part am I missing here?

Answer 1

You're creating a "hot" Observable so it's emitting even when there are no subscribers.

When creating Observables with new Observable() you can optionally return so called teardown (or dispose) function that should cleanup any resources if needed.

So in your case you'll want to stop the timer.

new Observable(subscriber => {
  ...
  const handler = setTimeout(() => {...});

  return () => clearTimeout(handler);
});

Or if you have multiple timers you'd call clearTimeout on all of them.

Answer 2

Which part am I missing here?

I think the part you're missing is that it's the responsibility of the observable to respect the contract: if it's being asked to stop emitting, it should stop emitting. So your observable should do the following:

const digestor$ = new Observable(subscriber => {
  let keepGoing = true;
  for (let i = 0; i < 4711 && keepGoing; i++) {
    setTimeout(() => {
      if (keepGoing) {
        const hash = createHashWithNLeadingZeroes(i);
        subscriber.next(hash);
      }
    }, i);

    return () => keepGoing = false; // this function is called when the subscriber unsubscribes
  }
});

It's usually a better idea to rely on existing factory functions and operators to achieve the desired behavior. You could do the equivalent of the above observable by using range(), timer() and map(), for example.