feat(core): implement takeUntilDestroyed in rxjs-interop (#49154)

This commit implements an RxJS operator `takeUntilDestroyed` which
terminates an Observable when the current context (component, directive,
etc) is destroyed. `takeUntilDestroyed` will inject the current `DestroyRef`
if none is provided, or use one provided as an argument.

PR Close #49154

goldens/public-api/core/rxjs-interop/index.md

@@ -4,7 +4,9 @@
 
 ```ts
 
+import { DestroyRef } from '@angular/core';
 import { Injector } from '@angular/core';
+import { MonoTypeOperatorFunction } from 'rxjs';
 import { Observable } from 'rxjs';
 import { Signal } from '@angular/core';
 
@@ -22,6 +24,9 @@ export interface FromSignalOptions {
     injector?: Injector;
 }
 
+// @public
+export function takeUntilDestroyed<T>(destroyRef?: DestroyRef): MonoTypeOperatorFunction<T>;
+
 // (No @packageDocumentation comment for this package)
 
 ```

packages/core/rxjs-interop/src/index.ts

@@ -8,3 +8,4 @@
 
 export {fromObservable} from './from_observable';
 export {fromSignal, FromSignalOptions} from './from_signal';
+export {takeUntilDestroyed} from './take_until_destroyed';

packages/core/rxjs-interop/src/take_until_destroyed.ts

@@ -0,0 +1,36 @@
+/**
+ * @license
+ * Copyright Google LLC All Rights Reserved.
+ *
+ * Use of this source code is governed by an MIT-style license that can be
+ * found in the LICENSE file at https://angular.io/license
+ */
+
+import {assertInInjectionContext, DestroyRef, inject} from '@angular/core';
+import {MonoTypeOperatorFunction, Observable} from 'rxjs';
+import {takeUntil} from 'rxjs/operators';
+
+/**
+ * Operator which completes the Observable when the calling context (component, directive, service,
+ * etc) is destroyed.
+ *
+ * @param destroyRef optionally, the `DestroyRef` representing the current context. This can be
+ *     passed explicitly to use `takeUntilDestroyed` outside of an injection context. Otherwise, the
+ * current `DestroyRef` is injected.
+ *
+ * @developerPreview
+ */
+export function takeUntilDestroyed<T>(destroyRef?: DestroyRef): MonoTypeOperatorFunction<T> {
+  if (!destroyRef) {
+    assertInInjectionContext(takeUntilDestroyed);
+    destroyRef = inject(DestroyRef);
+  }
+
+  const destroyed$ = new Observable<void>(observer => {
+    destroyRef!.onDestroy(observer.next.bind(observer));
+  });
+
+  return <T>(source: Observable<T>) => {
+    return source.pipe(takeUntil(destroyed$));
+  };
+}

packages/core/rxjs-interop/test/take_until_destroyed_spec.ts

@@ -0,0 +1,66 @@
+/**
+ * @license
+ * Copyright Google LLC All Rights Reserved.
+ *
+ * Use of this source code is governed by an MIT-style license that can be
+ * found in the LICENSE file at https://angular.io/license
+ */
+
+import {DestroyRef, EnvironmentInjector, Injector, runInInjectionContext} from '@angular/core';
+import {BehaviorSubject} from 'rxjs';
+
+import {takeUntilDestroyed} from '../src/take_until_destroyed';
+
+describe('takeUntilDestroyed', () => {
+  it('should complete an observable when the current context is destroyed', () => {
+    const injector = Injector.create({providers: []}) as EnvironmentInjector;
+    const source$ = new BehaviorSubject(0);
+    const tied$ = runInInjectionContext(injector, () => source$.pipe(takeUntilDestroyed()));
+
+    let completed = false;
+    let last = 0;
+
+    tied$.subscribe({
+      next(value) {
+        last = value;
+      },
+      complete() {
+        completed = true;
+      }
+    });
+
+    source$.next(1);
+    expect(last).toBe(1);
+
+    injector.destroy();
+    expect(completed).toBeTrue();
+    source$.next(2);
+    expect(last).toBe(1);
+  });
+
+  it('should allow a manual DestroyRef to be passed', () => {
+    const injector = Injector.create({providers: []}) as EnvironmentInjector;
+    const source$ = new BehaviorSubject(0);
+    const tied$ = source$.pipe(takeUntilDestroyed(injector.get(DestroyRef)));
+
+    let completed = false;
+    let last = 0;
+
+    tied$.subscribe({
+      next(value) {
+        last = value;
+      },
+      complete() {
+        completed = true;
+      }
+    });
+
+    source$.next(1);
+    expect(last).toBe(1);
+
+    injector.destroy();
+    expect(completed).toBeTrue();
+    source$.next(2);
+    expect(last).toBe(1);
+  });
+});