refactor: reduce RxJS dependency

In this commit, we reduce the usage of RxJS utilities as they may be redundant. For example,
we replace the replay subject with a behavior subject and remove some unnecessary operators.

CHANGELOG.md

@@ -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)

packages/hmr-plugin/src/internal/hmr-lifecycle.ts

@@ -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

packages/store/internals/src/ngxs-app-bootstrapped-state.ts

@@ -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();
   }
 }

packages/store/src/internal/lifecycle-state-manager.ts

@@ -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,

packages/websocket-plugin/src/websocket-handler.ts

@@ -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);
       });