How to best handle multiple subscriptions to the same observable on the template?

Question

Let's say I have an observable called 'todoList$'. Using the 'async' operator, I'm able to automatically subscribe/unsubscribe from it. The problem in the code below is that there are two identical subscriptions to the same observable:

<ng-container *ngIf="(todoList$ | async).length > 0>
  <div *ngFor="let todo of todoList$ | async">
    ...

This isn't very DRY, and consequentially, we allocate memory for a subscription that could handled more efficiently.

Because of the syntax in the ngIf condition, I don't believe I can use the 'as' keyword to create a template variable for the observable output. Instead what works is when I use the RxJs 'share' operator from the component file:

todoList$ = this.store.select(todoList).pipe(tap(x => {console.log('testing')}), share());
//testing  

Without the share operator, "testing" is printed twice. This leads me to believe the share() operator solves this problem. If it does, not exactly sure why/how? Since this can be a prevalent issue/ code smell, what's the best way of handling multiple subscriptions that are identical within the same template?

I acknowledge there are a few flavors of a similar question floating on StackOverflow. But none have given me exactly what I'm looking for.

Answer 1

Another option (I think it's more simple)

<ng-container *ngIf="todoList$|async as todoList;else loading">
    <div *ngFor="let todo of todoList">
        {{todo}}
    </div>
  <div *ngIf="!todoList.length">Empty</div>
</ng-container>
<ng-template #loading>loading...</ng-template>

Another one, using an intermediate object (*)

<ng-container *ngIf="{data:todoList$|async} as todoList">
  <div *ngIf="!todoList.data">loading...</div>
    <div *ngFor="let todo of todoList.data">
        {{todo}}
    </div>
  <div *ngIf="!todoList.data.length">Empty</div>
</ng-container>

(*) See that the first *ngIf return always true, but under the ng-container we has in todoList.data the data.

Answer 2

As a general rule I use the shareReplay({ refCount: true, bufferSize: 1 }) operator at the end of every Observable inside my template. I also add it to base observables which I use to branch of other observables which are then used in the template. This will make sure subscriptions are shared among every subscriber, and by using the shareReplay you can get the last emitted result as well inside your component by using take(1).

The reason of the { refCount: true, bufferSize: 1 } is that if you just use shareReplay(1) it can cause leaking subscriptions, regardless if you are using the async pipe.

Back to your example, the answer provided by Michael D is not bad, and it makes sense to do it that way. However, it does require some logic in the template, which I personally frown upon.

So, as long as you are using shareReplay, there is really no downside into using multiple async calls in your template, and you can even make them descriptive and re-usable throughout your template by defining them in your component:

export class TodoComponent {
  readonly todoList$ = this.store.select(todoList).pipe(
    shareReplay({ refCount: true, bufferSize: 1 })
  );

  readonly hasTodos$ = this.todoList$.pipe(
    map((todos) => todos?.length > 0),
    shareReplay({ refCount: true, bufferSize: 1 })
  );
}

You can then keep your template descriptive:

<ng-container *ngIf="hasTodos$ | async>
  <div *ngFor="let todo of todoList$ | async">
  <!-- -->

_{don't forget your trackBy!}

---

If you dislike repeating your code, you can even create your custom operator, and use that:

export function shareRef<T>() {
  return (source: Observable<T>) => source.pipe(
    shareReplay({ refCount: true, bufferSize: 1 })
  );
}

Which changes your observable to:

readonly todoList$ = this.store.select(todoList).pipe(
  shareRef()
);

Answer 3

If it does, not exactly sure why/how?

Let's see how share() is defined:

function shareSubjectFactory() {
  return new Subject<any>();
}

return (source: Observable<T>) => refCount()(multicast(shareSubjectFactory)(source)) as Observable<T>;

First of all,

(source: Observable<T>) => refCount()(multicast(shareSubjectFactory)(source))

is the same as

(source: Observable<T>) => source.pipe(
  multicast(shareSubjectFactory),
  refCount()
)

multicast will return a ConnectableObservable, which is still an Observable, but, among other things, it exposes a connect method.

// Inside `multicast` operator

const connectable: any = Object.create(source, connectableObservableDescriptor);
connectable.source = source;
connectable.subjectFactory = subjectFactory;

return <ConnectableObservable<R>> connectable;

Source

Another interesting thing about it is that when subscribed to, the subscriber will be added to the Subject's list of subscribers and the main source will not be subscribed until connect is called:

_subscribe(subscriber: Subscriber<T>) {
  return this.getSubject().subscribe(subscriber);
}

protected getSubject(): Subject<T> {
  const subject = this._subject;
  if (!subject || subject.isStopped) {
    this._subject = this.subjectFactory();
  }
  return this._subject!;
}

For example:

const src$ = privateSrc.pipe(
  tap(() => console.log('from src')),
  share(),
  tap(() => console.log('from share()')),
)

When src$ is subscribed:

// Subscriber #1
src$.subscribe(/* ... */)

the subscriber will be added to the Subject's subscribers list and the source src$, will be subscribed. Why? Because share also uses refCount, which subscribes to the source if a new subscriber is registered when there were no previous active subscribers and will unsubscribe from the source if there are no more active subscribers.

Let's take a look at another example:

const src$ = (new Observable(s => {
  console.warn('[SOURCE] SUBSCRIBED')

  setTimeout(() => {
    s.next(1);
  }, 1000);
})).pipe(share());

// First subscriber,
// because it's the first one, `refCount` will to its job and the source will be subscribed
// and this subscriber will be added to the `Subject`'s subscribers list
// note that the source sends the data asynchronously
src$.subscribe(/* ... */)

// The second subscriber
// since the source is already subscribed, `refCount` won't subscribe to it again
// instead, this new subscriber will be added to `Subject`'s list
src$.subscribe(/* ... */)

After 1s, the source will send the value 1 and the subject will receive that value and will send it to its registered subscribers.

This is how refCount does its magic:

// When a new subscriber is registered

(<any> connectable)._refCount++;

// `RefCountSubscriber` will make sure that if no more subscribers are left
// the source will be unsubscribed
const refCounter = new RefCountSubscriber(subscriber, connectable);

// Add the subscriber to the `Subject`'s list
const subscription = connectable.subscribe(refCounter);

if (!refCounter.closed) {
  (<any> refCounter).connection = connectable.connect();
}

return subscription;

And ConnectableObservable.connect is defined as follows:

connect(): Subscription {
  let connection = this._connection;
  
  if (!connection) {
    // If the source wasn't subscribed before

    this._isComplete = false;
    connection = this._connection = new Subscription();
    
    // Subscribing to the source
    // Every notification send by the source will be first received by `Subject`
    connection.add(this.source
      .subscribe(new ConnectableSubscriber(this.getSubject(), this)));
    
    /* ... */
  }
  return connection;
}

---

So, if we have a src$ observable that needs to be subscribed multiple times in the template, we can apply the above mentioned concepts.

However, there is an important aspect that we should be aware of.

If our template looks like this:

<!-- #1 -->
<div *ngIf="src$ | async"></div>

<!-- ... -->

<!-- #2 -->
<div *ngIf="src$ | async"></div>

and src$:

src$ = store.pipe(select(/* ... */), share())

then, if store already has the value, it will be retrieved synchronously, meaning that when #1 will be registered, the store will be subscribed and will send that value, but notice that at that time #2 is not yet subscribed, so it won't receive anything.

If the source is asynchronous, then we should have no problems, since the subscriptions in the template will most likely by synchronous.

But, when the source is synchronous, you could solve this problem this way:

src$ = store.pipe(
  select(/* ... */),
  subscribeOn(asyncScheduler),
  share()
)

subscribeOn(asyncScheduler) is roughly the same as delaying the subscription of the source with setTimeout(() => {}, 0). But, this allows for #2 to be subscribed so that when the source is finally subscribed, both subscribers will receive that value.

Answer 4

use *ngIf with this type of condition. I hope it will help you.

<ng-container *ngIf="(todoList$ | async) as todoList">
  <ng-container *ngIf="todoList && todoList != undefined && todoList.length">
    <div *ngFor="let todo of todoList">
      ...

Answer 5

You could actually still use the as signature in the *ngIf directive to have only one active subscription. Try the following

<ng-container *ngIf="(todoList$ | async) as todoList">
  <ng-container *ngIf="todoList.length > 0">
    <div *ngFor="let todo of todoList">
      ...