Yes, that’s expected behavior! The reason lies in how Angular’s change detection mechanism operates under the hood. To truly grasp the inner workings of the async pipe, let’s build our own version from scratch! We’ll name it SubscribePipe to distinguish it from the original implementation. Our custom pipe must satisfy these requirements:
  • Handle Observables (we'll skip Promises in this discussion)
  • Avoid duplicate subscriptions
  • Integrate seamlessly with OnPush change detection
  • Prevent memory leaks
We intend to use it in templates like this: 
@Component({
  selector: 'my-app',
  template: `
    <div *ngIf="show">
      Value: {{ obs$ | subscribe }}
    </div>
  `,
  standalone: true,
  imports: [CommonModule, SubscribePipe],
  // changeDetection: ChangeDetectionStrategy.OnPush // <-- after we handle it we should uncomment this
})
export class AppComponent {
  show = true;

  obs$ = interval(500).pipe(
    tap((x) => {
      if (x === 10) {
        this.show = false;
      }
    })
  );
}
Enter fullscreen mode Exit fullscreen mode

Constructing a standalone pipe 

import { Pipe, PipeTransform } from '@angular/core';

@Pipe({
  name: 'subscribe',
  standalone: true
})
export class SubscribePipe implements PipeTransform {
  transform() {}
}
Enter fullscreen mode Exit fullscreen mode
Our pipe needs to accept an Observable type, which means it must handle Observables, Subjects, BehaviorSubjects, and ReplaySubjects. Let’s define a Subscribable type that encompasses all these categories (as a generic type). 
type Subscribable<T> = Observable<T> | Subject<T> | BehaviorSubject<T> | ReplaySubject<T>;
Enter fullscreen mode Exit fullscreen mode
With that type in place, let’s refactor the pipe to use it! Since we want to infer the type of the subscribable, we'll make the pipe class generic. Additionally, our pipe should gracefully accept undefined or null as inputs. 
export class SubscribePipe<T> implements PipeTransform {
  transform(obs: Subscribable<T> | null | undefined) {}
}
Enter fullscreen mode Exit fullscreen mode

Managing subscriptions

First, we must verify that the observable isn't null or undefined. If it is, we return null. 
transform(obs: Subscribable<T> | null): T | null {
  if (!obs) {
    return null;
  }
}
Enter fullscreen mode Exit fullscreen mode
Next, let’s subscribe to the observable, retain its most recent emission, and return that value. 
latestValue: T | null = null;

transform(obs: Subscribable<T> | null): T | null {
  if (!obs) {
    return null;
  }

  obs.subscribe(value => {
    this.latestValue = value;
  });

  return this.latestValue;
}
Enter fullscreen mode Exit fullscreen mode
This approach fails! Why? Because each time change detection runs, the pipe inspects the arguments in the transform method. If they remain unchanged, it simply returns the previously cached value. I covered this topic in more detail in a prior article: It’s ok to use function calls in Angular templates!, which explains how pipes implement memoization and how you can replicate that behavior with regular functions. This is precisely where we need to disable pipe memoization by setting the pure: false flag. 
@Pipe({
  name: 'subscribe',
  standalone: true,
  pure: false // <-- It is true by default
})
Enter fullscreen mode Exit fullscreen mode
By setting the pure option to false, we are informing Angular that we will manage the memoization of the transform method ourselves. If we execute this code, we'll observe the following behavior: Not working solution This occurs because every time change detection is triggered, the pipe subscribes to the observable anew, while leaving the previous subscription active in memory. This inevitably leads to a memory leak! How do we resolve this? A simple reference comparison does the trick. We'll store the current observable reference. Each time the transform method gets called, we'll compare it against the stored one. If they match, we simply return the latest cached value. 
private currentObs: Subscribable<T> | null = null;

transform(obs: Subscribable<T> | null): T | null {
  if (!obs) {
    return null;
  }

  if (obs === this.currentObs) { // <-- simple equality check
    return this.latestValue;
  } else {
    this.currentObs = obs; // <-- save current observable to a class field

    obs.subscribe((value) => {
      this.latestValue = value;
    });
  }

  return this.latestValue;
}
Enter fullscreen mode Exit fullscreen mode
Upon checking the app now, everything appears functional! But we are not done yet. Working solution The issue persists: we're still creating a memory leak because we never unsubscribe from the observable. Let’s fix that!

Handling unsubscriptions

In the same way we kept track of latestValue and currentObs, we'll persist the active subscription and connect the observable's subscription to it.

private sub: Subscription | null = null;

transform(obs: Subscribable<T> | null): T | null {
  ...
  this.sub = obs.subscribe((value) => {
    this.latestValue = value;
  });
  ...
}
Enter fullscreen mode Exit fullscreen mode

We'll need to call unsubscribe in ngOnDestroy, but that's not the only place 💡. Unsubscribing is also required whenever we point to a different observable or assign null to the reference.

Consider this scenario:

@Component({
  selector: 'my-app',
  template: `
    <div *ngIf="show">{{ obs$ | subscribe }}</div>
  `,
  standalone: true,
  imports: [CommonModule, SubscribePipe],
})
export class AppComponent {
  show = true;

  ngOnInit() {
    setTimeout(() => {
      this.obs$ = of(20000);
    }, 2000);

    setTimeout(() => {
      this.obs$ = null;
    }, 4000);
  }

  obs$ = interval(500).pipe(
    tap((x) => {
      if (x === 10) {
        this.show = false;
      }
    })
  );
}
Enter fullscreen mode Exit fullscreen mode

We must dispose of the subscription in these situations too — otherwise, we're introducing a memory leak!

To keep things DRY, let's extract the unsubscription logic into a dispose() method.

private dispose() {
  if (this.sub) { // <-- first we check if we have a subscription
    this.sub.unsubscribe(); // <-- unsubscribe from the observable
    this.sub = null; // <-- remove the subscription reference
  }
}
Enter fullscreen mode Exit fullscreen mode

Now let's put this method to work.

First, hook it into ngOnDestroy(), and then apply it to the other situations we talked about.

ngOnDestroy() {
  this.dispose();
}
Enter fullscreen mode Exit fullscreen mode 
transform(obs: Subscribable<T> | null): T | null {
  if (!obs) {
    this.dispose(); // <-- if we have a current sub and change the obs to be null we need to dispose it
    return null;
  }

  if (obs === this.currentObs) {
    return this.latestValue;
  } else {
    this.dispose(); // <-- before subscribing to a new observable, we need to dispose the existing one

    this.currentObs = obs;

    this.sub = obs.subscribe((value) => {
      this.latestValue = value;
    });
  }

  return this.latestValue;
}
Enter fullscreen mode Exit fullscreen mode

The app continues to operate correctly — now without the risk of memory leaks 🎉.

Are we wrapped up? Not quite. Our pipe is still incompatible with OnPush ChangeDetection.

Experiment on your end: switch your component to changeDetection: ChangeDetectionStrategy.OnPush and watch the app go blank.

The remedy is the standard fix for change detection issues 😈😄: call cdr.markForCheck() right after updating the value.

private cdr = inject(ChangeDetectorRef); // <-- inject CDRef here

transform(obs: Subscribable<T> | null): T | null {
  ...
  this.sub = obs.subscribe((value) => {
    this.latestValue = value;
    this.cdr.markForCheck(); // <-- mark the component as dirty here, after we have updated the latestValue
  });
  ...
}
Enter fullscreen mode Exit fullscreen mode

That does the trick!

Refining the code

Let's streamline everything: extract the subscription logic into a dedicated method (and include error throwing), null out cdr and currentObs in ngOnDestroy (clearing those references helps avoid leaks), and tidy up the transform method for better readability.

private subscribe(obs: Subscribable<T>) {
  this.currentObs = obs;

  this.sub = obs.subscribe({
    next: (res) => {
      this.latestValue = res;
      this.cdr.markForCheck();
    },
    error: (error) => {
      throw error;
    },
  });
}
Enter fullscreen mode Exit fullscreen mode 
transform(obs: Subscribable<T> | null): T | null {
  if (!obs) {
    this.dispose();
    return null;
  }

  // here we check if the obs are not the same instead of checking if they are the same
  if (obs !== this.currentObs) {
    this.dispose();
    this.subscribe(obs); // <-- use the method we extracted above
  }

  return this.latestValue;
}
Enter fullscreen mode Exit fullscreen mode 
ngOnDestroy() {
  this.dispose();
  this.cdr = null;
  this.currentObs = null;
}
Enter fullscreen mode Exit fullscreen mode

Wrapping up

You can find the complete source code here.

We've now covered nearly every scenario Angular's async pipe handles — promises being the only exception, and those are simple to support as well. We've also clarified why Angular's async pipe is not pure and why that's perfectly fine.

I trust you found this write-up valuable and picked up a thing or two along the way.

--

I'm quite active on Twitter about Angular — sharing the latest news, videos, podcasts, updates, RFCs, pull requests, and more. If that interests you, follow me at @Enea_Jahollari. And if this article helped you, consider following me on Dev.to for similar content.

Thanks for sticking around!