- Broadcast service events without connected sockets via Redis

- Document base websocket server class

Author: oklemenz2

CHANGELOG.md

@@ -5,11 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
-## Version 0.1.3 - 2024-xx-xx
+## Version 0.1.3 - 2024-01-11
 
 ### Fixed
 
-- tbd
+- Broadcast service events without connected sockets via Redis
+- Document base websocket server class
 
 ## Version 0.1.2 - 2024-01-08
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cap-js-community/websocket",
-  "version": "0.1.3",
+  "version": "0.2.0",
   "description": "WebSocket adapter for CDS",
   "homepage": "https://cap.cloud.sap/",
   "engines": {

src/adapter/redis.js

@@ -6,38 +6,43 @@ const cds = require("@sap/cds");
 const LOG = cds.log("websocket/redis");
 
 class RedisAdapter {
-  constructor(server, channel, options) {
+  constructor(server, prefix, options) {
     this.server = server;
-    this.channel = channel;
+    this.prefix = prefix;
     this.options = options;
   }
 
   async setup() {
     this.client = await redis.createPrimaryClientAndConnect();
   }
 
-  async on() {
+  async on(service) {
     if (!this.client) {
       return;
     }
     try {
-      await this.client.subscribe(this.channel);
-      this.client.on("message", (channel, message) => {
-        if (channel === this.channel) {
-          this.server.wss.broadcastAll(message);
+      const channel = this.prefix + service;
+      await this.client.subscribe(channel, async (message, messageChannel) => {
+        try {
+          if (messageChannel === channel) {
+            await this.server.broadcast(service, message);
+          }
+        } catch (err) {
+          LOG?.error(err);
         }
       });
     } catch (err) {
       LOG?.error(err);
     }
   }
 
-  async emit(message) {
+  async emit(service, message) {
     if (!this.client) {
       return;
     }
     try {
-      await this.client.publish(this.channel, message);
+      const channel = this.prefix + service;
+      await this.client.publish(channel, message);
     } catch (err) {
       LOG?.error(err);
     }

src/index.js

@@ -108,13 +108,13 @@ function serveWebSocketService(socketServer, service, options) {
     if (["websocket", "ws"].includes(endpoint.kind)) {
       const servicePath = normalizeServicePath(endpoint.path, options.path);
       try {
+        bindServiceEvents(socketServer, servicePath, service);
         socketServer.service(servicePath, (socket) => {
           try {
             socket.setup();
             bindServiceDefaults(socket, service);
             bindServiceOperations(socket, service);
             bindServiceEntities(socket, service);
-            bindServiceEvents(socket, service);
             emitConnect(socket, service);
           } catch (err) {
             LOG?.error(err);
@@ -128,9 +128,16 @@ function serveWebSocketService(socketServer, service, options) {
   }
 }
 
-async function emitConnect(socket, service) {
-  if (service.operations(WebSocketAction.Connect).length) {
-    await processEvent(socket, service, undefined, WebSocketAction.Connect);
+function bindServiceEvents(socketServer, servicePath, service) {
+  for (const event of service.events()) {
+    service.on(event, async (req) => {
+      const localEventName = serviceLocalName(service, event.name);
+      try {
+        await socketServer.broadcast(servicePath, localEventName, req.data, null, true);
+      } catch (err) {
+        LOG?.error(err);
+      }
+    });
   }
 }
 
@@ -190,20 +197,9 @@ function bindServiceEntities(socket, service) {
   }
 }
 
-function bindServiceEvents(socket, service) {
-  for (const event of service.events()) {
-    service.on(event, async (req) => {
-      const localEventName = serviceLocalName(service, event.name);
-      await processEmit(socket, service, localEventName, req.data);
-    });
-  }
-}
-
-async function processEmit(socket, service, event, data) {
-  try {
-    socket.emit(event, data);
-  } catch (err) {
-    LOG?.error(err);
+async function emitConnect(socket, service) {
+  if (service.operations(WebSocketAction.Connect).length) {
+    await processEvent(socket, service, undefined, WebSocketAction.Connect);
   }
 }
 

src/socket/base.js

@@ -4,19 +4,37 @@ const cds = require("@sap/cds");
 const cookie = require("cookie");
 const crypto = require("crypto");
 
+/**
+ * Base class for a websocket server
+ */
 class SocketServer {
+  /**
+   * Constructor for websocket server
+   * @param server HTTP server from express app
+   * @param path Protocol path, e.g. '/ws'
+   */
   constructor(server, path) {
     this.id = crypto.randomUUID();
     this.server = server;
     this.path = path;
     cds.ws = null;
   }
 
+  /**
+   * Setup websocket server with async operations
+   * @returns {Promise<void>} Promise when setup is completed
+   */
   async setup() {}
 
+  /**
+   * Connect a service to websocket
+   * @param service service path, e.g. "/chat"
+   * @param connected Callback function to be called on every websocket connection passing socket functions (i.e. ws.on("connection", connected))
+   */
   service(service, connected) {
     connected &&
       connected({
+        service,
         socket: null,
         setup: () => {},
         context: () => {},
@@ -27,6 +45,21 @@ class SocketServer {
       });
   }
 
+  /**
+   * Broadcast to all websocket clients
+   * @param service service path, e.g. "/chat"
+   * @param event Event name
+   * @param data Data object
+   * @param socket Broadcast client to be excluded
+   * @param multiple Broadcast across multiple websocket servers
+   * @returns {Promise<void>} Promise when broadcasting completed
+   */
+  async broadcast(service, event, data, socket, multiple) {}
+
+  /**
+   * Mock the HTTP response object and make available at request.res
+   * @param request HTTP request
+   */
   static mockResponse(request) {
     // Mock response (not available in websocket, CDS middlewares need it)
     const res = request.res ?? {};
@@ -67,6 +100,10 @@ class SocketServer {
     request.res = res;
   }
 
+  /**
+   * Apply the authorization cookie to authorization header for local authorization testing in mocked auth scenario
+   * @param request HTTP request
+   */
   static applyAuthCookie(request) {
     // Apply cookie to authorization header
     if (["mocked"].includes(cds.env.requires.auth.kind) && !request.headers.authorization && request.headers.cookie) {

src/socket/socket.io.js

@@ -36,6 +36,7 @@ class SocketIOServer extends SocketServer {
           DEBUG?.("Disconnected", socket.id);
         });
         const serviceSocket = {
+          service,
           socket,
           setup: () => {
             this._enforceAuth(socket);
@@ -69,6 +70,15 @@ class SocketIOServer extends SocketServer {
     });
   }
 
+  async broadcast(service, event, data, socket, multiple) {
+    if (socket) {
+      socket.broadcast.emit(event, data);
+    } else {
+      const io = this.io.of(service);
+      io.emit(event, data);
+    }
+  }
+
   async _applyAdapter() {
     try {
       const adapterImpl = cds.env.requires?.websocket?.adapter?.impl;

src/socket/ws.js

@@ -11,30 +11,17 @@ class SocketWSServer extends SocketServer {
   constructor(server, path) {
     super(server, path);
     this.wss = new WebSocket.Server({ server });
-    this.wss.broadcast = (message, socket) => {
-      this.wss.clients.forEach((client) => {
-        if (client !== socket && client.readyState === WebSocket.OPEN) {
-          client.send(message);
-        }
-      });
-    };
-    this.wss.broadcastAll = (message) => {
-      this.wss.clients.forEach((client) => {
-        if (client.readyState === WebSocket.OPEN) {
-          client.send(message);
-        }
-      });
-    };
-    this.adapter = null;
     cds.ws = this.wss;
     cds.wss = this.wss;
+    this.adapter = null;
   }
 
   async setup() {
     await this._applyAdapter();
   }
 
   service(service, connected) {
+    this.adapter?.on(service);
     this.wss.on("connection", async (ws, request) => {
       ws.request = request;
       if (ws.request?.url !== `${this.path}${service}`) {
@@ -51,6 +38,7 @@ class SocketWSServer extends SocketServer {
         try {
           connected &&
             connected({
+              service,
               socket: ws,
               setup: () => {
                 this._enforceAuth(ws);
@@ -66,18 +54,16 @@ class SocketWSServer extends SocketServer {
               },
               on: (event, callback) => {
                 ws.on("message", async (message) => {
-                  let payload;
+                  let payload = {};
                   try {
                     payload = JSON.parse(message);
                   } catch (_) {
                     // ignore
                   }
                   try {
                     if (payload?.event === event) {
+                      await this.adapter?.emit(service, message);
                       await callback(payload.data);
-                      if (this.adapter) {
-                        await this.adapter.emit(message);
-                      }
                     }
                   } catch (err) {
                     LOG?.error(err);
@@ -93,13 +79,7 @@ class SocketWSServer extends SocketServer {
                 );
               },
               broadcast: (event, data) => {
-                this.wss.broadcast(
-                  JSON.stringify({
-                    event,
-                    data,
-                  }),
-                  ws,
-                );
+                this.broadcast(service, event, data, ws, true);
               },
               disconnect() {
                 ws.disconnect();
@@ -112,6 +92,28 @@ class SocketWSServer extends SocketServer {
     });
   }
 
+  async broadcast(service, event, data, socket, multiple) {
+    const clients = [];
+    this.wss.clients.forEach((client) => {
+      if (
+        client.readyState === WebSocket.OPEN &&
+        client !== socket &&
+        client.request?.url === `${this.path}${service}`
+      ) {
+        clients.push(client);
+      }
+    });
+    if (clients.length > 0 || multiple) {
+      const message = !data ? event : JSON.stringify({ event, data });
+      clients.forEach((client) => {
+        client.send(message);
+      });
+      if (multiple) {
+        await this.adapter?.emit(service, message);
+      }
+    }
+  }
+
   async _applyAdapter() {
     try {
       const adapterImpl = cds.env.requires?.websocket?.adapter?.impl;
@@ -120,11 +122,10 @@ class SocketWSServer extends SocketServer {
         if (cds.env.requires.websocket?.adapter?.options) {
           options = { ...options, ...cds.env.requires.websocket?.adapter?.options };
         }
-        const channel = options?.key ?? "websocket";
+        const prefix = options?.key ?? "websocket";
         this.adapterFactory = require(`../adapter/${adapterImpl}`);
-        this.adapter = new this.adapterFactory(this, channel, options);
-        await this.adapter.setup();
-        await this.adapter.on();
+        this.adapter = new this.adapterFactory(this, prefix, options);
+        await this.adapter?.setup();
       }
     } catch (err) {
       LOG?.error(err);

test/base.test.js

@@ -19,6 +19,7 @@ describe("Base", () => {
     socketServer.setup();
     socketServer.service("test", connected);
     expect(socket).toEqual({
+      service: "test",
       socket: null,
       setup: expect.any(Function),
       context: expect.any(Function),

test/redis_ws.test.js

@@ -39,7 +39,8 @@ describe("Redis", () => {
     expect(redis.createClient).toHaveBeenCalledWith({ url: "uri" });
     expect(redis.client.connect).toHaveBeenCalledWith();
     expect(redis.client.on).toHaveBeenNthCalledWith(1, "error", expect.any(Function));
-    expect(redis.client.on).toHaveBeenNthCalledWith(2, "message", expect.any(Function));
-    expect(redis.client.publish).toHaveBeenCalledWith("websocket", expect.any(Buffer));
+    expect(redis.client.subscribe).toHaveBeenNthCalledWith(1, "websocket/chat", expect.any(Function));
+    expect(redis.client.subscribe).toHaveBeenNthCalledWith(2, "websocket/main", expect.any(Function));
+    expect(redis.client.publish).toHaveBeenCalledWith("websocket/chat", expect.any(Buffer));
   });
 });