Unsubscribing from a ReplaySubject returned asObservable();

Question

Header

If you are essentially having the same question and your context is Angular you may want to read all the comments in the answer for more context.

Short version of this question

When doing let observe$ = someReplaySubject.asObservable(), are we no longer required to unsubscribe from observe$? In other words can we call let observe$ = someReplaySubject.asObservable() from multiple Angular component instances and not worry about the connection from the notify instance to the corresponding Observable?

Assumptions

If we have a ReplaySubject<Todo[]> instance (Lets call it rxTodo and we subscribe to it from Angular components by doing:

let todoObs$:Observable<Todo[]> = rxTodo$.asObservable() then even if the component is destroyed the todoObs$ references created in each component will dangle until the component itself is destroyed.

Background

I'm attempting a Store API for Angular and I have a replay subject that broadcasts changes to slices of the store. Here is the method that allows subscriptions to happen (The notifyCount tracks subscriptions, so that if there are none we don't bother with notification):

  /**
   * Subscribe to receive slice updates.
   * @example
     <pre>
    let todos$ = slice.subscribe();
    </pre>
  */
  public subscribe(): Observable<E[]> {
    this.notifyCount++;
    return this.notify.asObservable();
  }

Above I'm attempting to follow the recommended best practice of returning an Observable instead of from the ReplaySubject.

This is the corresponding unsubscribe method:

  /**
   * Unsubscribe from slice updates.
   * 
   * @example 
    <pre>
      slice.unsubscribe(o);
    </pre>
  */
  public unsubscribe(o: ReplaySubject<E[]>) {
    o.unsubscribe();
    this.notifyCount--;
  }

I had to make the o argument a ReplaySubject type in order to unsubscribe. However that conflicts with the Observable type that the subscribe method returns.

When attempting to test like this:

incompleteSlice.unsubscribe(incomplete$);

The message returned is this:

[ts] Argument of type 'Observable' is not assignable to parameter of type 'ReplaySubject'. Property 'scheduler' is missing in type 'Observable'. let incomplete$: Observable

Any thoughts on how to fix this?

Update

One obvious thought that just came to mind is that perhaps returning asObservable means that we no longer need to actually unsubscribe from that observable. We can just leave it dangling in the event that the Angular component is destroyed?

Answer 1

I think you are confusing unsubscribing from Subscription and from Subject.

What is actually recommended in Angular is unsubscribing from any open Subscription when the component gets destroyed. A Subscription is returned from Observable.subscribe. When it gets unsubscribed, it doesn't receive values from the source observable anymore, which is what you want in most cases.

Unsubscribing from Subject has a different effect. It switches the Subject into a closed state where you can no longer call next or subscribe to it. You can look directly at the source code to understand whats going on.

Unsubscribing from an Observable is not possible, that's in part why you get the error.

In order to follow the recommendation, you should either make a list of all Subscriptions, that is, adding each Subscription to a list that you can access later, like that:

this.subscriptions$.push(obs$.subscribe...))

And then on component destroy, call:

this.subscriptions$.forEach(sub -> sub.unsubscribe());

But even better and recommended by Core Developer Ben Lesh in this article is not imperatively calling unsubscribe, but instead utilizing the takeUntil operator.

There is a reference implementation on StackOverflow for using this pattern with Angular, that you can use as a starting point.