rxState implementation

package.json

@@ -20,7 +20,9 @@
     "@angular/platform-browser": "^17.0.0-rc.0",
     "@angular/platform-browser-dynamic": "^17.0.0-rc.0",
     "@angular/router": "^17.0.0-rc.0",
+    "@rx-angular/state": "^16.3.0",
     "firebase": "^10.4.0",
+    "ngxtension": "^0.14.1",
     "rxfire": "^6.0.5",
     "rxjs": "~7.8.0",
     "tslib": "^2.3.0",

src/app/app.component.ts

@@ -6,7 +6,6 @@ import { RouterOutlet } from '@angular/router';
   selector: 'app-root',
   standalone: true,
   imports: [CommonModule, RouterOutlet],
-  template: ` <router-outlet></router-outlet> `,
-  styles: [],
+  template: ` <router-outlet /> `,
 })
 export class AppComponent {}

src/app/home/home.component.spec.ts

@@ -18,9 +18,7 @@ describe('HomeComponent', () => {
         {
           provide: MessageService,
           useValue: {
-            add$: {
-              next: jest.fn(),
-            },
+            add: () => {},
             messages: jest.fn().mockReturnValue(mockMessages),
           },
         },
@@ -66,7 +64,7 @@ describe('HomeComponent', () => {
         const testValue = 'hello';
         messageInput.triggerEventHandler('send', testValue);
 
-        expect(messageService.add$.next).toHaveBeenCalledWith(testValue);
+        expect(messageService.actions.add).toHaveBeenCalledWith(testValue);
       });
     });
   });

src/app/home/home.component.ts

@@ -23,7 +23,7 @@ import { Router } from '@angular/router';
         [messages]="messageService.messages()"
         [activeUser]="authService.user()"
       />
-      <app-message-input (send)="messageService.add$.next($event)" />
+      <app-message-input (send)="messageService.actions.add($event)" />
     </div>
   `,
   imports: [

src/app/shared/data-access/auth.service.ts

@@ -1,4 +1,4 @@
-import { Injectable, computed, inject, signal } from '@angular/core';
+import { Injectable, inject } from '@angular/core';
 import { from, defer } from 'rxjs';
 import {
   User,
@@ -9,39 +9,24 @@ import {
 import { authState } from 'rxfire/auth';
 import { Credentials } from '../interfaces/credentials';
 import { AUTH } from 'src/app/app.config';
-import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
+import {rxState} from "@rx-angular/state";
 
 export type AuthUser = User | null | undefined;
 
 interface AuthState {
   user: AuthUser;
 }
 
-@Injectable({
-  providedIn: 'root',
-})
+@Injectable({providedIn: 'root'})
 export class AuthService {
   private auth = inject(AUTH);
 
-  // sources
-  private user$ = authState(this.auth);
-
-  // state
-  private state = signal<AuthState>({
-    user: undefined,
+  private state = rxState<AuthState>(({set, connect}) => {
+    set({user: null});
+    connect('user', authState(this.auth));
   });
 
-  // selectors
-  user = computed(() => this.state().user);
-
-  constructor() {
-    this.user$.pipe(takeUntilDestroyed()).subscribe((user) =>
-      this.state.update((state) => ({
-        ...state,
-        user,
-      }))
-    );
-  }
+  user = this.state.signal('user');
 
   login(credentials: Credentials) {
     return from(

src/app/shared/data-access/message.service.spec.ts

@@ -37,7 +37,7 @@ xdescribe('MessageService', () => {
         content: 'test',
       };
 
-      service.add$.next(testMessage.content);
+      service.actions.add(testMessage.content);
 
       expect(mockCollection).toHaveBeenCalledWith({} as any, 'messages');
       expect(mockAddDoc).toHaveBeenCalledWith(mockCollectionReference as any, {

src/app/shared/data-access/message.service.ts

@@ -1,81 +1,60 @@
-import { Injectable, computed, inject, signal } from '@angular/core';
-import { takeUntilDestroyed, toObservable } from '@angular/core/rxjs-interop';
-import { EMPTY, Observable, Subject, defer, exhaustMap, from } from 'rxjs';
-import { collection, query, orderBy, limit, addDoc } from 'firebase/firestore';
-import { collectionData } from 'rxfire/firestore';
-import { catchError, filter, map, retry } from 'rxjs/operators';
+import {inject, Injectable} from '@angular/core';
+import {defer, exhaustMap, Observable, of} from 'rxjs';
+import {addDoc, collection, limit, orderBy, query} from 'firebase/firestore';
+import {collectionData} from 'rxfire/firestore';
+import {catchError, filter, ignoreElements, map, retry} from 'rxjs/operators';
 
-import { FIRESTORE } from 'src/app/app.config';
-import { Message } from '../interfaces/message';
-import { AuthService } from './auth.service';
+import {FIRESTORE} from 'src/app/app.config';
+import {Message} from '../interfaces/message';
+import {AuthService, AuthUser} from './auth.service';
+import {rxState} from "@rx-angular/state";
+import {rxActions} from "@rx-angular/state/actions";
+import {rxEffects} from "@rx-angular/state/effects";
 
 interface MessageState {
+  authUser: AuthUser;
   messages: Message[];
   error: string | null;
 }
 
-@Injectable({
-  providedIn: 'root',
-})
+@Injectable({providedIn: 'root',})
 export class MessageService {
   private firestore = inject(FIRESTORE);
   private authService = inject(AuthService);
-  private authUser$ = toObservable(this.authService.user);
 
-  // sources
-  messages$ = this.getMessages().pipe(
-    // restart stream when user reauthenticates
-    retry({
-      delay: () => this.authUser$.pipe(filter((user) => !!user)),
-    })
-  );
-  add$ = new Subject<Message['content']>();
-  error$ = new Subject<string>();
-  logout$ = this.authUser$.pipe(filter((user) => !user));
+  actions = rxActions<{ add: Message['content'] }>();
 
-  // state
-  private state = signal<MessageState>({
-    messages: [],
-    error: null,
-  });
+  state = rxState<MessageState>(({set, connect, select}) => {
+    set({ messages: [], error: null });
+    connect('authUser', this.authService.user);
 
-  // selectors
-  messages = computed(() => this.state().messages);
-  error = computed(() => this.state().error);
+    const authenticated$ = select('authUser').pipe(map((user) => !!user));
 
-  constructor() {
-    // reducers
-    this.messages$.pipe(takeUntilDestroyed()).subscribe((messages) =>
-      this.state.update((state) => ({
-        ...state,
-        messages,
-      }))
-    );
+    connect('messages', this.getMessages().pipe(
+      retry({delay: () => authenticated$ }) // restart stream when user re-authenticates
+    ));
+
+    connect('messages', authenticated$.pipe(filter(x => !x)), () => []);
 
-    this.add$
-      .pipe(
-        takeUntilDestroyed(),
-        exhaustMap((message) => this.addMessage(message))
+    connect(
+      'error',
+      this.actions.add$.pipe(
+        exhaustMap((message) => this.addMessage(message)),
+        ignoreElements(),
+        catchError(err => of(err))
       )
-      .subscribe({
-        error: (err) => {
-          console.log(err);
-          this.error$.next('Failed to send message');
-        },
-      });
+    );
+  });
 
-    this.logout$
-      .pipe(takeUntilDestroyed())
-      .subscribe(() =>
-        this.state.update((state) => ({ ...state, messages: [] }))
-      );
+  effects = rxEffects(({ register }) => {
+    register(this.state.select('messages'), (messages) => {
+      console.log('messages', messages);
+    })
+  })
 
-    this.error$
-      .pipe(takeUntilDestroyed())
-      .subscribe((error) =>
-        this.state.update((state) => ({ ...state, error }))
-      );
-  }
+  // selectors
+  messages = this.state.signal('messages');
+  error = this.state.signal('error');
 
   private getMessages() {
     const messagesCollection = query(