SYSTEST-9338 - Send events to the WS object that subscribed to it

server/src/events.mjs

@@ -49,11 +49,20 @@ function logFatalErr() {
 const eventListenerMap = {};
 
 // Associate this message ID with this method so if/when events are sent, we know which message ID to use
-function registerEventListener(userId, oMsg) {
-  if ( ! eventListenerMap[userId] ) {
+function registerEventListener(userId, oMsg, ws) {
+  if (!eventListenerMap[userId]) {
     eventListenerMap[userId] = {};
   }
-  eventListenerMap[userId][oMsg.method] = oMsg.id;
+
+  if (!eventListenerMap[userId][oMsg.method]) {
+    eventListenerMap[userId][oMsg.method] = { id: oMsg.id, wsArr: [] };
+  }
+
+  // Check if ws is already in the wsArr before pushing
+  if (!eventListenerMap[userId][oMsg.method].wsArr.includes(ws)) {
+    eventListenerMap[userId][oMsg.method].wsArr.push(ws);
+  }
+
   logger.debug(`Registered event listener mapping: ${userId}:${oMsg.method}:${oMsg.id}`);
 }
 
@@ -80,12 +89,23 @@ function getRegisteredEventListener(userId, method) {
 
 // Remove mapping from event listener request method name from our map
 // Attempts to send events to this listener going forward will fail
-function deregisterEventListener(userId, oMsg) {
-  if ( ! eventListenerMap[userId] ) { return; }
-  delete eventListenerMap[userId][oMsg.method];
-  logger.debug(`Deregistered event listener for method: ${userId}:${oMsg.method}`);
-}
+function deregisterEventListener(userId, oMsg, ws) {
+  if (!eventListenerMap[userId] || !eventListenerMap[userId][oMsg.method]) {
+    return;
+  }
 
+  const wsArr = eventListenerMap[userId][oMsg.method].wsArr;
+  const wsIndex = wsArr.findIndex((item) => item === ws);
+
+  if (wsIndex !== -1) {
+    wsArr.splice(wsIndex, 1);
+    logger.debug(`Deregistered event listener mapping: ${userId}:${oMsg.method}`);
+  }
+
+  if (wsArr.length === 0) {
+    delete eventListenerMap[userId][oMsg.method];
+  }
+}
 // Is the given (incoming) message one that enables or disables an event listener?
 // Example: {"jsonrpc":"2.0","method":"lifecycle.onInactive","params":{"listen":true|false},"id":1}
 // Key: The 'on' in the unqualified method name, and (2) the params.listen parameter (regardless of true|false)
@@ -142,17 +162,24 @@ function sendBroadcastEvent(ws, userId, method, result, msg, fSuccess, fErr, fFa
 }
 
 // sending response to web-socket
-function emitResponse(ws, finalResult, msg, userId, method){
-  let id = getRegisteredEventListener(userId, method);
+function emitResponse(finalResult, msg, userId, method) {
+  const listener = getRegisteredEventListener(userId, method);
+  if (!listener) {
+    logger.debug('Event message could not be sent because a listener was not found');
+    return;
+  }
+  const { id, wsArr } = listener;
   const oEventMessage = {
     jsonrpc: '2.0',
     id: id,
     result: finalResult
   };
   const eventMessage = JSON.stringify(oEventMessage);
-  // Could do, but why?: const dly = stateManagement.getAppropriateDelay(user, method); await util.delay(dly);
-  ws.send(eventMessage);
-  logger.info(`${msg}: Sent event message to user ${userId}: ${eventMessage}`);
+
+  wsArr.forEach((ws) => {
+    ws.send(eventMessage);
+    logger.info(`${msg}: Sent event message to user ${userId}: ${eventMessage}`);
+  });
 }
 
 // sendEvent to handle post API event calls, including pre- and post- event trigger processing
@@ -247,7 +274,7 @@ function coreSendEvent(isBroadcast, ws, userId, method, result, msg, fSuccess, f
         // looping over each web-sockets of same group
         if ( wsUserMap && wsUserMap.size >=1 ) {
           wsUserMap.forEach ((userWithSameGroup, ww) => {
-            emitResponse(ww, finalResult, msg, userWithSameGroup, method);
+            emitResponse(finalResult, msg, userWithSameGroup, method);
           });
           fSuccess.call(null);
         } else {
@@ -256,7 +283,7 @@ function coreSendEvent(isBroadcast, ws, userId, method, result, msg, fSuccess, f
           throw new Error(msg);
         }
       } else {
-        emitResponse(ws, finalResult, msg, userId, method);
+        emitResponse(finalResult, msg, userId, method);
         fSuccess.call(null);
       }
     }

server/src/messageHandler.mjs

@@ -83,14 +83,14 @@ async function handleMessage(message, userId, ws) {
 
   if ( events.isEventListenerOnMessage(oMsg) ) {
     events.sendEventListenerAck(userId, ws, oMsg);
-    events.registerEventListener(userId, oMsg);
+    events.registerEventListener(userId, oMsg, ws);
     return;
   }
 
   // Handle JSON-RPC messages that are event listener disable requests
 
   if ( events.isEventListenerOffMessage(oMsg) ) {
-    events.deregisterEventListener(userId, oMsg);
+    events.deregisterEventListener(userId, oMsg, ws);
     return;
   }
 

server/test/suite/events.test.mjs

@@ -31,7 +31,8 @@ test(`events.registerEventListener works properly`, () => {
     method: "",
     id: 12,
   };
-  events.registerEventListener("12345", dummyObject);
+  const dummyWebSocket = { send: () => {} };
+  events.registerEventListener("12345", dummyObject, dummyWebSocket);
   expect(spy).toHaveBeenCalled();
 });
 
@@ -92,7 +93,8 @@ test(`events.getRegisteredEventListener works properly if path`, () => {
 test(`events.deregisterEventListener works properly`, () => {
   const spy = jest.spyOn(logger, "debug");
   const oMsgdummy = { method: "lifecycle.onInactive", id: 12 };
-  events.deregisterEventListener(oMsgdummy);
+  const dummyWebSocket = { send: () => {} };
+  events.deregisterEventListener("12345", oMsgdummy.method, dummyWebSocket);
   expect(spy).toHaveBeenCalled();
 });
 
@@ -177,7 +179,8 @@ test(`events.isEventListenerOffMessage works properly`, () => {
 test(`events.sendEventListenerAck works properly`, () => {
   const spy = jest.spyOn(logger, "debug");
   const oMsgdummy = { method: "lifecycle.onInactive", id: 12 };
-  events.sendEventListenerAck('12345', { send: () => {} }, oMsgdummy);
+  const dummyWebSocket = { send: () => {} };
+  events.sendEventListenerAck("12345", dummyWebSocket, oMsgdummy);
   expect(spy).toHaveBeenCalled();
 });
 
@@ -205,9 +208,11 @@ test(`events.sendEvent works properly`, () => {
       call: () => {},
     },
   };
-  events.registerEventListener("12345", dummyObject);
+  const dummyWebSocket = { send: () => {} };
+
+  events.registerEventListener("12345", dummyObject, dummyWebSocket);
   events.sendEvent(
-    { send: () => {} },
+    dummyWebSocket,
     "12345",
     methodName,
     result,
@@ -286,8 +291,9 @@ test(`events.isAnyRegisteredInGroup works properly with if path`, () => {
 });
 
 test(`events.sendBroadcastEvent works properly`, () => {
+  const dummyWebSocket = { send: () => {} };
   const result = events.testExports.sendBroadcastEvent(
-    { send: () => {} },
+    dummyWebSocket,
     "12345",
     "core",
     {},
@@ -301,12 +307,15 @@ test(`events.sendBroadcastEvent works properly`, () => {
 
 test(`events.emitResponse works properly`, () => {
   const spy = jest.spyOn(logger, "info");
-  events.testExports.emitResponse(
-    { send: () => {} },
-    {},
-    "test_msg",
-    "12345",
-    "core"
-  );
+  const listenerObject = {
+    method: "core",
+    id: 12,
+  };
+  const dummyWebSocket = { send: () => {} };
+
+  // Register the event listener first to simulate a real-world scenario
+  events.registerEventListener("12345", listenerObject, dummyWebSocket);
+
+  events.testExports.emitResponse({}, "test_msg", "12345", "core");
   expect(spy).toHaveBeenCalled();
 });