ngxs · arturovt · Jan 3, 2025 · Jan 3, 2025
@@ -7,6 +7,7 @@ $ npm install @ngxs/store@dev
 ### To become next patch version
 
 - Refactor: Replace `ngOnDestroy` with `DestroyRef` [#2289](https://github.com/ngxs/store/pull/2289)
+- Refactor: Reduce RxJS dependency [#2292](https://github.com/ngxs/store/pull/2292)
 - Fix(store): Add root store initializer guard [#2278](https://github.com/ngxs/store/pull/2278)
 - Fix(store): Reduce change detection cycles with pending tasks [#2280](https://github.com/ngxs/store/pull/2280)
 - Fix(store): Complete action results on destroy [#2282](https://github.com/ngxs/store/pull/2282)

@@ -54,7 +54,8 @@ export class HmrLifecycle<T extends Partial<NgxsHmrLifeCycle<S>>, S> {
 
     const state$: Observable<any> = this.context.store.select(state => state);
 
-    this.appBootstrappedState.subscribe(() => {
+    this.appBootstrappedState.subscribe(bootstrapped => {
+      if (!bootstrapped) return;
       let eventId: number;
       const storeEventId: Subscription = state$.subscribe(() => {
         // setTimeout used for zone detection after set hmr state

@@ -1,14 +1,19 @@
-import { Injectable } from '@angular/core';
-import { ReplaySubject } from 'rxjs';
+import { DestroyRef, inject, Injectable } from '@angular/core';
+import { BehaviorSubject } from 'rxjs';
 
 @Injectable({ providedIn: 'root' })
-export class ɵNgxsAppBootstrappedState extends ReplaySubject<boolean> {
+export class ɵNgxsAppBootstrappedState extends BehaviorSubject<boolean> {
   constructor() {
-    super(1);
+    super(false);
+
+    const destroyRef = inject(DestroyRef);
+    // Complete the subject once the root injector is destroyed to ensure
+    // there are no active subscribers that would receive events or perform
+    // any actions after the application is destroyed.
+    destroyRef.onDestroy(() => this.complete());
   }
 
   bootstrap(): void {
     this.next(true);
-    this.complete();
   }
 }
@@ -1,8 +1,7 @@
-import { DestroyRef, inject, Injectable } from '@angular/core';
+import { inject, Injectable } from '@angular/core';
 import { ɵNgxsAppBootstrappedState } from '@ngxs/store/internals';
 import { getValue, InitState, UpdateState } from '@ngxs/store/plugins';
-import { ReplaySubject } from 'rxjs';
-import { filter, mergeMap, pairwise, startWith, takeUntil, tap } from 'rxjs/operators';
+import { EMPTY, mergeMap, pairwise, startWith } from 'rxjs';
 
 import { Store } from '../store';
 import { StateContextFactory } from './state-context-factory';
@@ -18,14 +17,8 @@ export class LifecycleStateManager {
   private _stateContextFactory = inject(StateContextFactory);
   private _appBootstrappedState = inject(ɵNgxsAppBootstrappedState);
 
-  private readonly _destroy$ = new ReplaySubject<void>(1);
-
   private _initStateHasBeenDispatched?: boolean;
 
-  constructor() {
-    inject(DestroyRef).onDestroy(() => this._destroy$.next());
-  }
-
   ngxsBootstrap(
     action: InitState | UpdateState,
     results: StatesAndDefaults | undefined
@@ -46,27 +39,40 @@ export class LifecycleStateManager {
       }
     }
 
+    // It does not need to unsubscribe because it is completed when the
+    // root injector is destroyed.
     this._internalStateOperations
       .getRootStateOperations()
       .dispatch(action)
       .pipe(
-        filter(() => !!results),
-        tap(() => this._invokeInitOnStates(results!.states)),
-        mergeMap(() => this._appBootstrappedState),
-        filter(appBootstrapped => !!appBootstrapped),
-        takeUntil(this._destroy$)
+        mergeMap(() => {
+          // If no states are provided, we safely complete the stream
+          // and do not proceed further.
+          if (!results) {
+            return EMPTY;
+          }
+
+          this._invokeInitOnStates(results!.states);
+          return this._appBootstrappedState;
+        })
       )
-      .subscribe(() => this._invokeBootstrapOnStates(results!.states));
+      .subscribe(appBootstrapped => {
+        if (appBootstrapped) {
+          this._invokeBootstrapOnStates(results!.states);
+        }
+      });
   }
 
   private _invokeInitOnStates(mappedStores: MappedStore[]): void {
     for (const mappedStore of mappedStores) {
       const instance: NgxsLifeCycle = mappedStore.instance;
 
       if (instance.ngxsOnChanges) {
+        // It does not need to unsubscribe because it is completed when the
+        // root injector is destroyed.
         this._store
           .select(state => getValue(state, mappedStore.path))
-          .pipe(startWith(undefined), pairwise(), takeUntil(this._destroy$))
+          .pipe(startWith(undefined), pairwise())
           .subscribe(([previousValue, currentValue]) => {
             const change = new NgxsSimpleChange(
               previousValue,

@@ -1,7 +1,8 @@
 import { Injectable, NgZone, inject, DestroyRef } from '@angular/core';
+import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
 import { Actions, Store, ofActionDispatched } from '@ngxs/store';
 import { getValue } from '@ngxs/store/plugins';
-import { ReplaySubject, Subject, fromEvent, takeUntil } from 'rxjs';
+import { Subject, fromEvent, takeUntil } from 'rxjs';
 
 import {
   ConnectWebSocket,
@@ -29,33 +30,29 @@ export class WebSocketHandler {
 
   private readonly _typeKey = this._options.typeKey!;
 
-  private readonly _destroy$ = new ReplaySubject<void>(1);
+  private readonly _destroyRef = inject(DestroyRef);
 
   constructor() {
     this._setupActionsListeners();
 
-    const destroyRef = inject(DestroyRef);
-    destroyRef.onDestroy(() => {
-      this._closeConnection(/* forcelyCloseSocket */ true);
-      this._destroy$.next();
-    });
+    this._destroyRef.onDestroy(() => this._closeConnection(/* forcelyCloseSocket */ true));
   }
 
   private _setupActionsListeners(): void {
     this._actions$
-      .pipe(ofActionDispatched(ConnectWebSocket), takeUntil(this._destroy$))
+      .pipe(ofActionDispatched(ConnectWebSocket), takeUntilDestroyed(this._destroyRef))
       .subscribe(({ payload }) => {
         this.connect(payload);
       });
 
     this._actions$
-      .pipe(ofActionDispatched(DisconnectWebSocket), takeUntil(this._destroy$))
+      .pipe(ofActionDispatched(DisconnectWebSocket), takeUntilDestroyed(this._destroyRef))
       .subscribe(() => {
         this._disconnect(/* forcelyCloseSocket */ true);
       });
 
     this._actions$
-      .pipe(ofActionDispatched(SendWebSocketMessage), takeUntil(this._destroy$))
+      .pipe(ofActionDispatched(SendWebSocketMessage), takeUntilDestroyed(this._destroyRef))
       .subscribe(({ payload }) => {
         this.send(payload);
       });

@@ -44,16 +44,16 @@ describe('WebSocketHandler cleanup', () => {
       );
       const store = ngModuleRef.injector.get(Store);
       const webSocketHandler = ngModuleRef.injector.get(WebSocketHandler);
-      const nextSpy = jest.fn();
-      webSocketHandler['_destroy$'].subscribe(nextSpy);
+      // @ts-expect-error private property.
+      const spy = jest.spyOn(webSocketHandler, '_closeConnection');
 
       store.dispatch(new ConnectWebSocket());
 
       server.on('connection', () => {
         server.on('close', () => {
           try {
             // Assert
-            expect(nextSpy).toHaveBeenCalled();
+            expect(spy).toHaveBeenCalledWith(/* forcelyCloseSocket */ true);
           } finally {
             server.stop(done);
           }