Merge pull request #49 from AmadeusITGroup/refactor/websocket-buffer-cleanup

Refactor/websocket buffer cleanup

Author: adxdits

projects/ngrx-devtool-demo/src/app/book-list/book.service.ts

@@ -1,10 +1,18 @@
 import { HttpClient } from '@angular/common/http';
 import { inject, Injectable } from '@angular/core';
 
-import { Observable } from 'rxjs';
-import { map } from 'rxjs/operators';
+import { Observable, of } from 'rxjs';
+import { catchError, map } from 'rxjs/operators';
 import { Book } from './book.model';
 
+const FALLBACK_BOOKS: Book[] = [
+  { id: '1', volumeInfo: { title: 'Awakenings', authors: ['Oliver Sacks'] } },
+  { id: '2', volumeInfo: { title: 'The Man Who Mistook His Wife for a Hat', authors: ['Oliver Sacks'] } },
+  { id: '3', volumeInfo: { title: 'An Anthropologist on Mars', authors: ['Oliver Sacks'] } },
+  { id: '4', volumeInfo: { title: 'Musicophilia', authors: ['Oliver Sacks'] } },
+  { id: '5', volumeInfo: { title: 'The Island of the Colourblind', authors: ['Oliver Sacks'] } },
+];
+
 @Injectable({ providedIn: 'root' })
 export class GoogleBooksService {
   private readonly http = inject(HttpClient);
@@ -14,6 +22,9 @@ export class GoogleBooksService {
       .get<{ items: Book[] }>(
         'https://www.googleapis.com/books/v1/volumes?maxResults=5&orderBy=relevance&q=oliver%20sacks'
       )
-      .pipe(map((books) => books.items || []));
+      .pipe(
+        map((books) => books.items || []),
+        catchError(() => of(FALLBACK_BOOKS))
+      );
   }
 }

projects/ngrx-devtool/src/lib/core/actions-interceptor.service.ts

@@ -3,7 +3,7 @@ import { Actions } from '@ngrx/effects';
 import { Action } from '@ngrx/store';
 import { filter, Subject, takeUntil, tap } from 'rxjs';
 
-import { DevToolMessage, EffectEvent, TrackedAction } from './core.models';
+import { DEFAULT_WS_URL, DevToolMessage, EffectEvent, TrackedAction } from './core.models';
 import { EffectTrackerService } from './effect-tracker.service';
 import { WebSocketService, WebSocketMessage } from './websocket.service';
 
@@ -16,7 +16,7 @@ export class ActionsInterceptorService implements OnDestroy {
   private readonly destroy$ = new Subject<void>();
   private initialized = false;
 
-  initialize(wsUrl = 'ws://localhost:4000'): void {
+  initialize(wsUrl = DEFAULT_WS_URL): void {
     if (this.initialized) {
       return;
     }

projects/ngrx-devtool/src/lib/core/core.models.ts

@@ -1,5 +1,7 @@
 import { Action } from '@ngrx/store';
 
+export const DEFAULT_WS_URL = 'ws://localhost:4000';
+
 export type EffectLifecycle = 'triggered' | 'emitted' | 'executed' | 'error';
 
 export interface EffectEvent {

projects/ngrx-devtool/src/lib/core/websocket.service.ts

@@ -1,11 +1,14 @@
 import { inject, Injectable, OnDestroy, PLATFORM_ID } from '@angular/core';
 import { isPlatformBrowser } from '@angular/common';
 import { BehaviorSubject, Observable } from 'rxjs';
+import { DEFAULT_WS_URL } from './core.models';
 
 export interface WebSocketMessage {
   readonly type: string;
 }
 
+const MAX_BUFFER_SIZE = 200;
+
 @Injectable({ providedIn: 'root' })
 export class WebSocketService implements OnDestroy {
   private readonly platformId = inject(PLATFORM_ID);
@@ -31,7 +34,7 @@ export class WebSocketService implements OnDestroy {
     return this.incomingMessages$.asObservable();
   }
 
-  initialize(wsUrl = 'ws://localhost:4000'): void {
+  initialize(wsUrl = DEFAULT_WS_URL): void {
     if (this.initialized) {
       // If already initialized with same URL, skip
       if (this.wsUrl === wsUrl) {
@@ -47,19 +50,17 @@ export class WebSocketService implements OnDestroy {
   }
 
   send<T extends WebSocketMessage>(message: T): void {
-    const payload = JSON.stringify(message);
-
-    if (this.isConnected && this.socket?.readyState === WebSocket.OPEN) {
-      this.socket.send(payload);
-    } else if (this.isBrowser) {
-      this.messageBuffer.push(payload);
-    }
+    this.bufferOrSend(JSON.stringify(message));
   }
 
   sendRaw(payload: string): void {
+    this.bufferOrSend(payload);
+  }
+
+  private bufferOrSend(payload: string): void {
     if (this.isConnected && this.socket?.readyState === WebSocket.OPEN) {
       this.socket.send(payload);
-    } else if (this.isBrowser) {
+    } else if (this.isBrowser && this.messageBuffer.length < MAX_BUFFER_SIZE) {
       this.messageBuffer.push(payload);
     }
   }
@@ -73,6 +74,7 @@ export class WebSocketService implements OnDestroy {
   private close(): void {
     this.socket?.close();
     this.socket = null;
+    this.messageBuffer = [];
     this.connectionState$.next(false);
     this.initialized = false;
     this.wsUrl = null;
@@ -109,9 +111,11 @@ export class WebSocketService implements OnDestroy {
   }
 
   private flushBuffer(): void {
-    while (this.messageBuffer.length > 0 && this.isConnected) {
-      const message = this.messageBuffer.shift();
-      if (message && this.socket?.readyState === WebSocket.OPEN) {
+    const buffered = this.messageBuffer;
+    this.messageBuffer = [];
+
+    for (const message of buffered) {
+      if (this.isConnected && this.socket?.readyState === WebSocket.OPEN) {
         this.socket.send(message);
       }
     }

projects/ngrx-devtool/src/lib/store/meta-reducer.ts

@@ -2,6 +2,7 @@ import { ActionReducer, Action } from '@ngrx/store';
 import { inject } from '@angular/core';
 import { PerformanceTrackerService } from '../performance/performance-tracker.service';
 import { WebSocketService } from '../core/websocket.service';
+import { DEFAULT_WS_URL } from '../core/core.models';
 
 export interface RenderPerformanceData {
   readonly renderTime: number;
@@ -22,13 +23,13 @@ export interface DevToolMetaReducerConfig {
 }
 
 export function createDevToolMetaReducer(
-  wsUrlOrConfig: string | DevToolMetaReducerConfig = 'ws://localhost:4000'
+  wsUrlOrConfig: string | DevToolMetaReducerConfig = DEFAULT_WS_URL
 ) {
   const config: DevToolMetaReducerConfig = typeof wsUrlOrConfig === 'string'
     ? { wsUrl: wsUrlOrConfig }
     : wsUrlOrConfig;
 
-  const wsUrl = config.wsUrl ?? 'ws://localhost:4000';
+  const wsUrl = config.wsUrl ?? DEFAULT_WS_URL;
   const enablePerf = config.enablePerformanceTracking ?? true;
 
   return function devToolMetaReducer<State>(
@@ -85,5 +86,5 @@ export function createDevToolMetaReducer(
 export function loggerMetaReducer<State>(
   reducer: ActionReducer<State>
 ): ActionReducer<State> {
-  return createDevToolMetaReducer('ws://localhost:4000')(reducer);
+  return createDevToolMetaReducer(DEFAULT_WS_URL)(reducer);
 }

projects/ngrx-devtool/src/lib/store/provide-devtool.ts

@@ -2,6 +2,7 @@ import { APP_INITIALIZER, Provider } from '@angular/core';
 import { EffectSources } from '@ngrx/effects';
 import { ActionsInterceptorService } from '../core/actions-interceptor.service';
 import { DevToolsEffectSources } from '../core/devtools-effect-sources';
+import { DEFAULT_WS_URL } from '../core/core.models';
 
 export interface DevToolConfig {
   readonly wsUrl?: string;
@@ -13,7 +14,7 @@ export function provideNgrxDevTool(config: DevToolConfig = {}): Provider[] {
     {
       provide: APP_INITIALIZER,
       useFactory: (interceptor: ActionsInterceptorService) => () => {
-        interceptor.initialize(config.wsUrl ?? 'ws://localhost:4000');
+        interceptor.initialize(config.wsUrl ?? DEFAULT_WS_URL);
       },
       deps: [ActionsInterceptorService],
       multi: true,

projects/ngrx-devtool/src/lib/testing/websocket.service.spec.ts

@@ -151,6 +151,19 @@ describe('WebSocketService', () => {
       expect(messages[0].type).toBe('MSG_1');
       expect(messages[1].type).toBe('MSG_2');
     });
+
+    it('should drop messages once the buffer reaches max size (200)', () => {
+      service.initialize();
+      // Don't open the socket — all messages go to buffer
+
+      for (let i = 0; i < 250; i++) {
+        service.send({ type: `MSG_${i}` } as WebSocketMessage);
+      }
+
+      // Open and flush — should only have the first 200
+      getLastWs().simulateOpen();
+      expect(getLastWs().sentMessages).toHaveLength(200);
+    });
   });
 
   describe('sendRaw()', () => {