feat: introduce message preprocessor for ws adapter

integration/websockets/e2e/ws-gateway.spec.ts

@@ -218,6 +218,33 @@ describe('WebSocketGateway (WsAdapter)', () => {
     );
   });
 
+  it('should preprocess message', async () => {
+    const testingModule = await Test.createTestingModule({
+      providers: [ApplicationGateway],
+    }).compile();
+    app = testingModule.createNestApplication();
+
+    const wsAdapter = new WsAdapter(app);
+    wsAdapter.setMessagePreprocessor(data => ({
+      event: data[0],
+      data: data[1],
+    }));
+    app.useWebSocketAdapter(wsAdapter);
+    await app.listen(3000);
+
+    ws = new WebSocket('ws://localhost:8080');
+    await new Promise(resolve => ws.on('open', resolve));
+
+    ws.send(JSON.stringify(['push', { test: 'test' }]));
+    await new Promise<void>(resolve =>
+      ws.on('message', data => {
+        expect(JSON.parse(data).data.test).to.be.eql('test');
+        ws.close();
+        resolve();
+      }),
+    );
+  });
+
   afterEach(async function () {
     await app.close();
   });

packages/platform-ws/adapters/ws-adapter.ts

@@ -25,6 +25,9 @@ type HttpServerRegistryKey = number;
 type HttpServerRegistryEntry = any;
 type WsServerRegistryKey = number;
 type WsServerRegistryEntry = any[];
+type WsMessagePreprocessor = (
+  message: any,
+) => { event: string; data: any } | void;
 
 const UNDERLYING_HTTP_SERVER_PORT = 0;
 
@@ -41,6 +44,7 @@ export class WsAdapter extends AbstractWsAdapter {
     WsServerRegistryKey,
     WsServerRegistryEntry
   >();
+  protected messagePreprocessor: WsMessagePreprocessor = message => message;
 
   constructor(appOrHttpServer?: INestApplicationContext | any) {
     super(appOrHttpServer);
@@ -138,7 +142,10 @@ export class WsAdapter extends AbstractWsAdapter {
     transform: (data: any) => Observable<any>,
   ): Observable<any> {
     try {
-      const message = JSON.parse(buffer.data);
+      const message = this.messagePreprocessor(JSON.parse(buffer.data));
+      if (!message) {
+        return EMPTY;
+      }
       const messageHandler = handlersMap.get(message.event);
       const { callback } = messageHandler;
       return transform(callback(message.data, message.event));
@@ -179,6 +186,10 @@ export class WsAdapter extends AbstractWsAdapter {
     this.wsServersRegistry.clear();
   }
 
+  public setMessagePreprocessor(preprocessor: WsMessagePreprocessor) {
+    this.messagePreprocessor = preprocessor;
+  }
+
   protected ensureHttpServerExists(
     port: number,
     httpServer = http.createServer(),