revert: "refactor(store): drop ɵwrapObserverCalls (#2371)" (#2383)

This reverts commit 5a744a3.

Author: markwhitfeld

CHANGELOG.md

@@ -12,7 +12,6 @@ $ npm install @ngxs/store@dev
 - Fix(store): Stop contributing to stability once app is stable [#2379](https://github.com/ngxs/store/pull/2379)
 - Refactor(store): Clear `_states` on destroy to aid GC under high load [#2365](https://github.com/ngxs/store/pull/2365)
 - Refactor(store): Add `debugName` to `computed` signals in `selectSignal` [#2370](https://github.com/ngxs/store/pull/2370)
-- Refactor(store): Drop `ɵwrapObserverCalls` [#2371](https://github.com/ngxs/store/pull/2371)
 - Fix(storage-plugin): Guard against `engine` may be falsy [#2367](https://github.com/ngxs/store/pull/2367) [#2368](https://github.com/ngxs/store/pull/2368)
 - Peformance(storage-plugin): Replace closure-based action matcher with direct type comparison [#2369](https://github.com/ngxs/store/pull/2369)
 - Fix(router-plugin): Avoid redundant NGXS state updates for identical router snapshots [#2372](https://github.com/ngxs/store/pull/2372)

packages/store/internals/src/custom-rxjs-operators.ts

@@ -0,0 +1,21 @@
+import { type MonoTypeOperatorFunction, Observable } from 'rxjs';
+
+export function ɵwrapObserverCalls<TValue>(
+  invokeFn: (fn: () => void) => void
+): MonoTypeOperatorFunction<TValue> {
+  return (source: Observable<TValue>) => {
+    return new Observable<TValue>(subscriber => {
+      return source.subscribe({
+        next(value) {
+          invokeFn(() => subscriber.next(value));
+        },
+        error(error) {
+          invokeFn(() => subscriber.error(error));
+        },
+        complete() {
+          invokeFn(() => subscriber.complete());
+        }
+      });
+    });
+  };
+}

packages/store/internals/src/index.ts

@@ -6,6 +6,7 @@ export { ɵINITIAL_STATE_TOKEN, ɵInitialState } from './initial-state';
 export { ɵNgxsAppBootstrappedState } from './ngxs-app-bootstrapped-state';
 export { ɵNGXS_STATE_CONTEXT_FACTORY, ɵNGXS_STATE_FACTORY } from './internal-tokens';
 export { ɵOrderedSubject, ɵOrderedBehaviorSubject } from './custom-rxjs-subjects';
+export { ɵwrapObserverCalls } from './custom-rxjs-operators';
 export { ɵStateStream } from './state-stream';
 export { ɵhasOwnProperty, ɵdefineProperty } from './object-utils';
 export { ɵNgxsActionRegistry } from './action-registry';

packages/store/internals/src/state-stream.ts

@@ -2,6 +2,7 @@ import { DestroyRef, inject, Injectable, Signal, untracked } from '@angular/core
 import { toSignal } from '@angular/core/rxjs-interop';
 import { Observable } from 'rxjs';
 
+import { ɵwrapObserverCalls } from './custom-rxjs-operators';
 import { ɵOrderedBehaviorSubject } from './custom-rxjs-subjects';
 import { ɵPlainObject } from './symbols';
 
@@ -11,26 +12,10 @@ import { ɵPlainObject } from './symbols';
  */
 @Injectable({ providedIn: 'root' })
 export class ɵStateStream extends ɵOrderedBehaviorSubject<ɵPlainObject> {
-  readonly state: Signal<ɵPlainObject> = toSignal(
-    this.pipe(
-      source =>
-        new Observable(subscriber =>
-          source.subscribe({
-            // Without `untracked()`, if some signal happened to be
-            // read while computing the next state (e.g. reducers/selectors
-            // reading other signals before calling `stateStream.next()`),
-            // Angular would incorrectly record a dependency on that signal.
-            next: value => untracked(() => subscriber.next(value)),
-            error: error => untracked(() => subscriber.error(error)),
-            complete: () => untracked(() => subscriber.complete())
-          })
-        )
-    ),
-    {
-      manualCleanup: true,
-      requireSync: true
-    }
-  );
+  readonly state: Signal<ɵPlainObject> = toSignal(this.pipe(ɵwrapObserverCalls(untracked)), {
+    manualCleanup: true,
+    requireSync: true
+  });
 
   constructor() {
     super({});

packages/store/src/operators/leave-ngxs.ts

@@ -1,17 +1,11 @@
-import { Observable } from 'rxjs';
+import { ɵwrapObserverCalls } from '@ngxs/store/internals';
+
 import { InternalNgxsExecutionStrategy } from '../execution/execution-strategy';
 
 /**
  * Returns operator that will run
  * `subscribe` outside of the ngxs execution context
  */
 export function leaveNgxs<T>(ngxsExecutionStrategy: InternalNgxsExecutionStrategy) {
-  return (source: Observable<T>) =>
-    new Observable<T>(subscriber =>
-      source.subscribe({
-        next: value => ngxsExecutionStrategy.leave(() => subscriber.next(value)),
-        error: error => ngxsExecutionStrategy.leave(() => subscriber.error(error)),
-        complete: () => ngxsExecutionStrategy.leave(() => subscriber.complete())
-      })
-    );
+  return ɵwrapObserverCalls<T>(fn => ngxsExecutionStrategy.leave(fn));
 }