refactor(core): support an opt-in sync version of toObservable (#60640)

alxhub · thePunderWoman · commit 57a240b07e62 · 2025-04-01T15:56:39.000Z
This commit adds a flag `forceSyncFirstEmit` which opts in to the pending new behavior for `toObservable`, which emits the first value synchronously. This flag is only really meant for use during a short migration period while we update g3, and is not meant for prolonged usage. As a result, it's marked deprecated. PR Close #60640
diff --git a/goldens/public-api/core/rxjs-interop/index.api.md b/goldens/public-api/core/rxjs-interop/index.api.md
@@ -39,6 +39,8 @@ export function toObservable<T>(source: Signal<T>, options?: ToObservableOptions
 
 // @public
 export interface ToObservableOptions {
+    // @deprecated
+    forceSyncFirstEmit?: true;
     injector?: Injector;
 }
 
diff --git a/packages/core/rxjs-interop/src/to_observable.ts b/packages/core/rxjs-interop/src/to_observable.ts
@@ -15,6 +15,7 @@ import {
   Signal,
   untracked,
 } from '../../src/core';
+import {SIGNAL, ReactiveNode} from '../../primitives/signals';
 import {Observable, ReplaySubject} from 'rxjs';
 
 /**
@@ -30,6 +31,16 @@ export interface ToObservableOptions {
    * will be used.
    */
   injector?: Injector;
+
+  /**
+   * Temporary option for forcing a synchronous emit of the signal's initial value.
+   *
+   * This will eventually become the default behavior, but is opt-in to allow a short migration
+   * period.
+   *
+   * @deprecated will become default behavior
+   */
+  forceSyncFirstEmit?: true;
 }
 
 /**
@@ -42,6 +53,10 @@ export interface ToObservableOptions {
  * @developerPreview
  */
 export function toObservable<T>(source: Signal<T>, options?: ToObservableOptions): Observable<T> {
+  if (options?.forceSyncFirstEmit === true) {
+    return toObservableNext(source, options);
+  }
+
   !options?.injector && assertInInjectionContext(toObservable);
   const injector = options?.injector ?? inject(Injector);
   const subject = new ReplaySubject<T>(1);
@@ -67,3 +82,88 @@ export function toObservable<T>(source: Signal<T>, options?: ToObservableOptions
 
   return subject.asObservable();
 }
+
+/**
+ * New version of `toObservable` with always-synchronous first emit.
+ *
+ * This will eventually replace the other implementation.
+ */
+function toObservableNext<T>(source: Signal<T>, options?: ToObservableOptions): Observable<T> {
+  !options?.injector && assertInInjectionContext(toObservable);
+  const injector = options?.injector ?? inject(Injector);
+
+  return new Observable<T>((subscriber) => {
+    let firstVersion: number = -1;
+    let firstValue: T;
+    try {
+      firstValue = untracked(source);
+    } catch (err) {
+      // A failure on the first read just errors the observable without
+      // creating an effect.
+      subscriber.error(err);
+      return;
+    }
+    // We cache the `version` of the first value. This lets us avoid emitting
+    // this value a second time during the `effect`.
+    firstVersion = signalVersion(source);
+
+    // Emit the first value synchronously on subscription.
+    subscriber.next(firstValue);
+
+    // Create an effect that will watch the signal for future changes.
+    let firstEmit = true;
+    const ref = effect(
+      () => {
+        let value: T;
+        try {
+          // Read the value (& track it).
+          value = source();
+        } catch (err) {
+          // Errors cause the Observable stream to terminate.
+          untracked(() => subscriber.error(err));
+          cleanup(false);
+          return;
+        }
+
+        // Skip the emit of the value if it hasn't changed since the
+        // synchronous emit.
+        if (firstEmit) {
+          firstEmit = false;
+          if (signalVersion(source) === firstVersion) {
+            return;
+          }
+        }
+
+        untracked(() => subscriber.next(value));
+      },
+      {injector, manualCleanup: true},
+    );
+
+    const cleanup = (fromInjector: boolean) => {
+      ref.destroy();
+
+      if (fromInjector) {
+        // If we're cleaning up because the injector is destroyed, then our
+        // subscription is still active and we need to complete it.
+        subscriber.complete();
+      } else {
+        // Otherwise, remove the cleanup function. This both prevents the
+        // complete() event from being produced and allows memory to be
+        // reclaimed.
+        removeInjectorCleanupFn();
+      }
+    };
+
+    const removeInjectorCleanupFn = injector.get(DestroyRef).onDestroy(() => {
+      // Cleaning up from the `DestroyRef` means the stream is still active, so
+      // we should emit completion.
+      cleanup(true);
+    });
+
+    return () => cleanup(false);
+  });
+}
+
+function signalVersion(source: Signal<unknown>): number {
+  return (source[SIGNAL] as ReactiveNode).version;
+}
diff --git a/packages/core/rxjs-interop/test/to_observable_spec.ts b/packages/core/rxjs-interop/test/to_observable_spec.ts
@@ -176,4 +176,56 @@ describe('toObservable()', () => {
     flushEffects();
     expect(emits()).toBe(1);
   });
+
+  describe('synchronous emit', () => {
+    it('emits synchronously when requested', () => {
+      const counter = signal(0);
+      const log: number[] = [];
+      toObservable(counter, {injector, forceSyncFirstEmit: true}).subscribe((value) =>
+        log.push(value),
+      );
+
+      expect(log).toEqual([0]);
+
+      // Expect no emit after the effect.
+      flushEffects();
+      expect(log).toEqual([0]);
+
+      counter.set(1);
+      flushEffects();
+      expect(log).toEqual([0, 1]);
+    });
+
+    it('emits new values if they happen before the first effect run', () => {
+      const counter = signal(0);
+      const log: number[] = [];
+      toObservable(counter, {injector, forceSyncFirstEmit: true}).subscribe((value) =>
+        log.push(value),
+      );
+
+      expect(log).toEqual([0]);
+      counter.set(1);
+
+      // Expect new emit after the effect.
+      flushEffects();
+      expect(log).toEqual([0, 1]);
+    });
+
+    it('emits new values based on version, not identity', () => {
+      // Disable value equality checking.
+      const counter = signal(0, {equal: () => false});
+      const log: number[] = [];
+      toObservable(counter, {injector, forceSyncFirstEmit: true}).subscribe((value) =>
+        log.push(value),
+      );
+
+      expect(log).toEqual([0]);
+      // Because of the `equal` function, this will be treated as a new value.
+      counter.set(0);
+
+      // Expect new emit after the effect.
+      flushEffects();
+      expect(log).toEqual([0, 0]);
+    });
+  });
 });

Original file line number	Diff line number	Diff line change
`@@ -39,6 +39,8 @@ export function toObservable<T>(source: Signal<T>, options?: ToObservableOptions`
`39`	`39`
`40`	`40`	`// @public`
`41`	`41`	`export interface ToObservableOptions {`
	`42`	`+ // @deprecated`
	`43`	`+ forceSyncFirstEmit?: true;`
`42`	`44`	`injector?: Injector;`
`43`	`45`	`}`
`44`	`46`