CComms API updates

packages/node_modules/@node-red/runtime/lib/api/comms.js

@@ -36,7 +36,7 @@ var connections = [];
 const events = require("@node-red/util").events;
 
 function handleCommsEvent(event) {
-    publish(event.topic,event.data,event.retain);
+    publish(event.topic,event.data,event.retain,event.session,event.excludeSession);
 }
 function handleStatusEvent(event) {
     if (!event.status) {
@@ -74,13 +74,17 @@ function handleEventLog(event) {
     publish("event-log/"+event.id,event.payload||{});
 }
 
-function publish(topic,data,retain) {
+function publish(topic, data, retain, session, excludeSession) {
     if (retain) {
         retained[topic] = data;
     } else {
         delete retained[topic];
     }
-    connections.forEach(connection => connection.send(topic,data))
+    connections.forEach(connection => {
+        if ((!session || connection.session === session) && (!excludeSession || connection.session !== excludeSession)) {
+            connection.send(topic,data)
+        }
+    })
 }
 
 
@@ -109,6 +113,10 @@ var api = module.exports = {
     */
     addConnection: async function(opts) {
         connections.push(opts.client);
+        events.emit('comms:connection-added', {
+            session: opts.client.session,
+            user: opts.client.user
+        })
     },
 
     /**
@@ -126,6 +134,9 @@ var api = module.exports = {
                 break;
             }
         }
+        events.emit('comms:connection-removed', { 
+            session: opts.client.session
+        })
     },
 
     /**
@@ -157,5 +168,23 @@ var api = module.exports = {
     * @return {Promise<Object>} - resolves when complete
     * @memberof @node-red/runtime_comms
     */
-    unsubscribe: async function(opts) {}
+    unsubscribe: async function(opts) {},
+
+    /**
+     * @param {Object} opts
+     * @param {User} opts.user - the user calling the api
+     * @param {CommsConnection} opts.client - the client connection
+     * @param {String} opts.topic - the message topic
+     * @param {String} opts.data - the message data
+     * @return {Promise<Object>} - resolves when complete
+     */
+    receive: async function (opts) {
+        if (opts.topic) {
+            events.emit('comms:message:' + opts.topic, {
+                session: opts.client.session,
+                user: opts.user,
+                data: opts.data
+            })
+        }
+    }
 };