stream filtering by connection id

packages/api/src/controllers/serverConnections.js

@@ -152,7 +152,7 @@ module.exports = {
   },
 
   ping_meta: true,
-  async ping({ conidArray }) {
+  async ping({ conidArray, strmid }) {
     await Promise.all(
       _.uniq(conidArray).map(async conid => {
         const last = this.lastPinged[conid];
@@ -169,6 +169,7 @@ module.exports = {
         }
       })
     );
+    socket.setStreamIdFilter(strmid, { conid: conidArray });
     return { status: 'ok' };
   },
 

packages/api/src/main.js

@@ -77,6 +77,7 @@ function start() {
   }
 
   app.get(getExpressPath('/stream'), async function (req, res) {
+    const strmid = req.query.strmid;
     res.set({
       'Cache-Control': 'no-cache',
       'Content-Type': 'text/event-stream',
@@ -87,9 +88,9 @@ function start() {
 
     // Tell the client to retry every 10 seconds if connectivity is lost
     res.write('retry: 10000\n\n');
-    socket.addSseResponse(res);
+    socket.addSseResponse(res, strmid);
     onFinished(req, () => {
-      socket.removeSseResponse(res);
+      socket.removeSseResponse(strmid);
     });
   });
 

packages/api/src/utility/socket.js

@@ -1,7 +1,7 @@
 const _ = require('lodash');
 const stableStringify = require('json-stable-stringify');
 
-const sseResponses = [];
+const sseResponses = {};
 let electronSender = null;
 let pingConfigured = false;
 
@@ -12,12 +12,15 @@ module.exports = {
       pingConfigured = true;
     }
   },
-  addSseResponse(value) {
-    sseResponses.push(value);
+  addSseResponse(value, strmid) {
+    sseResponses[strmid] = {
+      response: value,
+      filter: {},
+    };
     this.ensurePing();
   },
-  removeSseResponse(value) {
-    _.remove(sseResponses, x => x == value);
+  removeSseResponse(strmid) {
+    delete sseResponses[strmid];
   },
   setElectronSender(value) {
     electronSender = value;
@@ -27,8 +30,23 @@ module.exports = {
     if (electronSender) {
       electronSender.send(message, data == null ? null : data);
     }
-    for (const res of sseResponses) {
-      res.write(`event: ${message}\ndata: ${stableStringify(data == null ? null : data)}\n\n`);
+    for (const strmid in sseResponses) {
+      let skipThisStream = false;
+      for (const key in sseResponses[strmid].filter) {
+        if (data && data[key]) {
+          if (!sseResponses[strmid].filter[key].includes(data[key])) {
+            skipThisStream = true;
+            break;
+          }
+        }
+      }
+      if (skipThisStream) {
+        continue;
+      }
+
+      sseResponses[strmid].response.write(
+        `event: ${message}\ndata: ${stableStringify(data == null ? null : data)}\n\n`
+      );
     }
   },
   emitChanged(key, params = undefined) {
@@ -36,4 +54,7 @@ module.exports = {
     this.emit('changed-cache', { key, ...params });
     // this.emit(key);
   },
+  setStreamIdFilter(strmid, filter) {
+    sseResponses[strmid].filter = filter;
+  },
 };