From f94a36613c2cd720b830524f9512ba1fc4ea3423 Mon Sep 17 00:00:00 2001
From: Nick O'Leary <nick.oleary@gmail.com>
Date: Mon, 23 Apr 2018 11:21:02 +0100
Subject: [PATCH] Split comms across api and runtime

---
 red/api/editor/comms.js              | 307 +++++++++++++--------------
 red/api/editor/index.js              |  23 +-
 red/api/index.js                     |   6 +-
 red/red.js                           |   3 +-
 red/runtime-api/comms.js             | 112 +++++++++-
 red/runtime-api/flows.js             |   1 -
 red/runtime-api/index.js             |  29 ++-
 red/runtime/nodes/registry/loader.js |  10 +-
 8 files changed, 307 insertions(+), 184 deletions(-)

diff --git a/red/api/editor/comms.js b/red/api/editor/comms.js
index d0e0cab2c..b7d2fea89 100644
--- a/red/api/editor/comms.js
+++ b/red/api/editor/comms.js
@@ -15,13 +15,17 @@
  **/
 
 var ws = require("ws");
-var log;
+
+var log = require("../../util").log; // TODO: separate module
+var Tokens;
+var Users;
+var Permissions;
 
 var server;
 var settings;
+var runtimeAPI;
 
 var wsServer;
-var pendingConnections = [];
 var activeConnections = [];
 
 var retained = {};
@@ -29,29 +33,145 @@ var retained = {};
 var heartbeatTimer;
 var lastSentTime;
 
-function handleStatus(event) {
-    publish("status/"+event.id,event.status,true);
-}
-function handleRuntimeEvent(event) {
-    log.trace("runtime event: "+JSON.stringify(event));
-    publish("notification/"+event.id,event.payload||{},event.retain);
-}
-function init(_server,runtime) {
+function init(_server,_settings,_runtimeAPI) {
     server = _server;
-    settings = runtime.settings;
-    log = runtime.log;
+    settings = _settings;
+    runtimeAPI = _runtimeAPI;
+    Tokens = require("../auth/tokens");
+    Users = require("../auth/users");
+    Permissions = require("../auth/permissions");
 
-    runtime.events.removeListener("node-status",handleStatus);
-    runtime.events.on("node-status",handleStatus);
+}
 
-    runtime.events.removeListener("runtime-event",handleRuntimeEvent);
-    runtime.events.on("runtime-event",handleRuntimeEvent);
+function generateSession(length) {
+    var c = "ABCDEFGHIJKLMNOPQRSTUZWXYZabcdefghijklmnopqrstuvwxyz1234567890";
+    var token = [];
+    for (var i=0;i<length;i++) {
+        token.push(c[Math.floor(Math.random()*c.length)]);
+    }
+    return token.join("");
+}
+
+function CommsConnection(ws) {
+    this.session = generateSession(32);
+    this.ws = ws;
+    this.stack = [];
+    this.user = null;
+    this.lastSentTime = 0;
+    var self = this;
+
+    log.audit({event: "comms.open"});
+    log.trace("comms.open "+self.session);
+    var pendingAuth = (settings.adminAuth != null);
+
+    if (!pendingAuth) {
+        addActiveConnection(self);
+    }
+    ws.on('close',function() {
+        log.audit({event: "comms.close",user:self.user, session: self.session});
+        log.trace("comms.close "+self.session);
+        removeActiveConnection(self);
+    });
+    ws.on('message', function(data,flags) {
+        var msg = null;
+        try {
+            msg = JSON.parse(data);
+        } catch(err) {
+            log.trace("comms received malformed message : "+err.toString());
+            return;
+        }
+        if (!pendingAuth) {
+            if (msg.subscribe) {
+                self.subscribe(msg.subscribe);
+                // handleRemoteSubscription(ws,msg.subscribe);
+            }
+        } else {
+            var completeConnection = function(userScope,sendAck) {
+                try {
+                    if (!userScope || !Permissions.hasPermission(userScope,"status.read")) {
+                        ws.send(JSON.stringify({auth:"fail"}));
+                        ws.close();
+                    } else {
+                        pendingAuth = false;
+                        addActiveConnection(self);
+                        if (sendAck) {
+                            ws.send(JSON.stringify({auth:"ok"}));
+                        }
+                    }
+                } catch(err) {
+                    console.log(err.stack);
+                    // Just in case the socket closes before we attempt
+                    // to send anything.
+                }
+            }
+            if (msg.auth) {
+                Tokens.get(msg.auth).then(function(client) {
+                    if (client) {
+                        Users.get(client.user).then(function(user) {
+                            if (user) {
+                                self.user = user;
+                                log.audit({event: "comms.auth",user:self.user});
+                                completeConnection(client.scope,true);
+                            } else {
+                                log.audit({event: "comms.auth.fail"});
+                                completeConnection(null,false);
+                            }
+                        });
+                    } else {
+                        log.audit({event: "comms.auth.fail"});
+                        completeConnection(null,false);
+                    }
+                });
+            } else {
+                if (anonymousUser) {
+                    log.audit({event: "comms.auth",user:anonymousUser});
+                    self.user = anonymousUser;
+                    completeConnection(anonymousUser.permissions,false);
+                    //TODO: duplicated code - pull non-auth message handling out
+                    if (msg.subscribe) {
+                        self.subscribe(msg.subscribe);
+                    }
+                } else {
+                    log.audit({event: "comms.auth.fail"});
+                    completeConnection(null,false);
+                }
+            }
+        }
+    });
+    ws.on('error', function(err) {
+        log.warn(log._("comms.error",{message:err.toString()}));
+    });
+}
+
+CommsConnection.prototype.send = function(topic,data) {
+    var self = this;
+    if (topic && data) {
+        this.stack.push({topic:topic,data:data});
+    }
+    if (!this._xmitTimer) {
+        this._xmitTimer = setTimeout(function() {
+            try {
+                self.ws.send(JSON.stringify(self.stack));
+                self.lastSentTime = Date.now();
+            } catch(err) {
+                removeActiveConnection(self);
+                log.warn(log._("comms.error-send",{message:err.toString()}));
+            }
+            delete self._xmitTimer;
+            self.stack = [];
+        },50);
+    }
+}
+
+CommsConnection.prototype.subscribe = function(topic) {
+    runtimeAPI.comms.subscribe({
+        user: this.user,
+        client: this,
+        topic: topic
+    })
 }
 
 function start() {
-    var Tokens = require("../auth/tokens");
-    var Users = require("../auth/users");
-    var Permissions = require("../auth/permissions");
     if (!settings.disableEditor) {
         Users.default().then(function(anonymousUser) {
             var webSocketKeepAliveTime = settings.webSocketKeepAliveTime || 15000;
@@ -68,90 +188,10 @@ function start() {
             });
 
             wsServer.on('connection',function(ws) {
-                log.audit({event: "comms.open"});
-                var pendingAuth = (settings.adminAuth != null);
-                ws._nr_stack = [];
-                ws._nr_ok2tx = true;
-
-                if (!pendingAuth) {
-                    activeConnections.push(ws);
-                } else {
-                    pendingConnections.push(ws);
-                }
-                ws.on('close',function() {
-                    log.audit({event: "comms.close",user:ws.user});
-                    removeActiveConnection(ws);
-                    removePendingConnection(ws);
-                });
-                ws.on('message', function(data,flags) {
-                    var msg = null;
-                    try {
-                        msg = JSON.parse(data);
-                    } catch(err) {
-                        log.trace("comms received malformed message : "+err.toString());
-                        return;
-                    }
-                    if (!pendingAuth) {
-                        if (msg.subscribe) {
-                            handleRemoteSubscription(ws,msg.subscribe);
-                        }
-                    } else {
-                        var completeConnection = function(userScope,sendAck) {
-                            try {
-                                if (!userScope || !Permissions.hasPermission(userScope,"status.read")) {
-                                    ws.send(JSON.stringify({auth:"fail"}));
-                                    ws.close();
-                                } else {
-                                    pendingAuth = false;
-                                    removePendingConnection(ws);
-                                    activeConnections.push(ws);
-                                    if (sendAck) {
-                                        ws.send(JSON.stringify({auth:"ok"}));
-                                    }
-                                }
-                            } catch(err) {
-                                // Just in case the socket closes before we attempt
-                                // to send anything.
-                            }
-                        }
-                        if (msg.auth) {
-                            Tokens.get(msg.auth).then(function(client) {
-                                if (client) {
-                                    Users.get(client.user).then(function(user) {
-                                        if (user) {
-                                            ws.user = user;
-                                            log.audit({event: "comms.auth",user:ws.user});
-                                            completeConnection(client.scope,true);
-                                        } else {
-                                            log.audit({event: "comms.auth.fail"});
-                                            completeConnection(null,false);
-                                        }
-                                    });
-                                } else {
-                                    log.audit({event: "comms.auth.fail"});
-                                    completeConnection(null,false);
-                                }
-                            });
-                        } else {
-                            if (anonymousUser) {
-                                log.audit({event: "comms.auth",user:anonymousUser});
-                                completeConnection(anonymousUser.permissions,false);
-                                //TODO: duplicated code - pull non-auth message handling out
-                                if (msg.subscribe) {
-                                    handleRemoteSubscription(ws,msg.subscribe);
-                                }
-                            } else {
-                                log.audit({event: "comms.auth.fail"});
-                                completeConnection(null,false);
-                            }
-                        }
-                    }
-                });
-                ws.on('error', function(err) {
-                    log.warn(log._("comms.error",{message:err.toString()}));
-                });
+                var commsConnection = new CommsConnection(ws);
             });
 
+
             wsServer.on('error', function(err) {
                 log.warn(log._("comms.error-server",{message:err.toString()}));
             });
@@ -161,7 +201,7 @@ function start() {
             heartbeatTimer = setInterval(function() {
                 var now = Date.now();
                 if (now-lastSentTime > webSocketKeepAliveTime) {
-                    publish("hb",lastSentTime);
+                    activeConnections.forEach(connection => connection.send("hb",lastSentTime));
                 }
             }, webSocketKeepAliveTime);
         });
@@ -179,63 +219,15 @@ function stop() {
     }
 }
 
-function publish(topic,data,retain) {
-    if (server) {
-        if (retain) {
-            retained[topic] = data;
-        } else {
-            delete retained[topic];
-        }
-        lastSentTime = Date.now();
-        activeConnections.forEach(function(conn) {
-            publishTo(conn,topic,data);
-        });
-    }
+function addActiveConnection(connection) {
+    activeConnections.push(connection);
+    runtimeAPI.comms.addConnection({client: connection});
 }
-
-function publishTo(ws,topic,data) {
-    if (topic && data) {
-        ws._nr_stack.push({topic:topic,data:data});
-    }
-    if (ws._nr_ok2tx && (ws._nr_stack.length > 0)) {
-        ws._nr_ok2tx = false;
-        try {
-            ws.send(JSON.stringify(ws._nr_stack));
-        } catch(err) {
-            removeActiveConnection(ws);
-            removePendingConnection(ws);
-            log.warn(log._("comms.error-send",{message:err.toString()}));
-        }
-        ws._nr_stack = [];
-        setTimeout(function() {
-            ws._nr_ok2tx = true;
-            publishTo(ws);
-        }, 50);  // TODO: OK so a 50mS update rate should prob not be hard-coded
-    }
-
-}
-
-function handleRemoteSubscription(ws,topic) {
-    var re = new RegExp("^"+topic.replace(/([\[\]\?\(\)\\\\$\^\*\.|])/g,"\\$1").replace(/\+/g,"[^/]+").replace(/\/#$/,"(\/.*)?")+"$");
-    for (var t in retained) {
-        if (re.test(t)) {
-            publishTo(ws,t,retained[t]);
-        }
-    }
-}
-
-function removeActiveConnection(ws) {
+function removeActiveConnection(connection) {
     for (var i=0;i<activeConnections.length;i++) {
-        if (activeConnections[i] === ws) {
+        if (activeConnections[i] === connection) {
             activeConnections.splice(i,1);
-            break;
-        }
-    }
-}
-function removePendingConnection(ws) {
-    for (var i=0;i<pendingConnections.length;i++) {
-        if (pendingConnections[i] === ws) {
-            pendingConnections.splice(i,1);
+            runtimeAPI.comms.removeConnection({client:connection})
             break;
         }
     }
@@ -244,6 +236,5 @@ function removePendingConnection(ws) {
 module.exports = {
     init:init,
     start:start,
-    stop:stop,
-    publish:publish
+    stop:stop
 }
diff --git a/red/api/editor/index.js b/red/api/editor/index.js
index a32c16ede..58e698d03 100644
--- a/red/api/editor/index.js
+++ b/red/api/editor/index.js
@@ -24,7 +24,6 @@ var info = require("./settings");
 var auth = require("../auth");
 var nodes = require("../admin/nodes"); // TODO: move /icons into here
 var needsPermission;
-var runtime;
 var runtimeAPI;
 var log = require("../../util").log; // TODO: separate module
 var i18n = require("../../util").i18n; // TODO: separate module
@@ -32,22 +31,23 @@ var i18n = require("../../util").i18n; // TODO: separate module
 var apiUtil = require("../util");
 
 var ensureRuntimeStarted = function(req,res,next) {
-    if (!runtime.isStarted()) {
-        log.error("Node-RED runtime not started");
-        res.status(503).send("Not started");
-    } else {
-        next();
-    }
+    runtimeAPI.isStarted().then( started => {
+        if (!started) {
+            log.error("Node-RED runtime not started");
+            res.status(503).send("Not started");
+        } else {
+            next()
+        }
+    })
 }
 
 module.exports = {
-    init: function(server, settings, _runtime, _runtimeAPI) {
-        runtime = _runtime;
+    init: function(server, settings, _runtimeAPI) {
         runtimeAPI = _runtimeAPI;
         needsPermission = auth.needsPermission;
         if (!settings.disableEditor) {
             info.init(runtimeAPI);
-            comms.init(server,runtime);
+            comms.init(server,settings,runtimeAPI);
 
             var ui = require("./ui");
 
@@ -122,6 +122,5 @@ module.exports = {
             comms.start();
         });
     },
-    stop: comms.stop,
-    publish: comms.publish
+    stop: comms.stop
 }
diff --git a/red/api/index.js b/red/api/index.js
index a46ba6968..298dacdd5 100644
--- a/red/api/index.js
+++ b/red/api/index.js
@@ -28,11 +28,11 @@ var adminApp;
 var server;
 var editor;
 
-function init(_server,settings,runtime,runtimeAPI) {
+function init(_server,settings,storage,runtimeAPI) {
     server = _server;
     if (settings.httpAdminRoot !== false) {
         adminApp = express();
-        auth.init(settings,runtime.storage);
+        auth.init(settings,storage);
 
         var maxApiRequestSize = settings.apiMaxLength || '5mb';
         adminApp.use(bodyParser.json({limit:maxApiRequestSize}));
@@ -57,7 +57,7 @@ function init(_server,settings,runtime,runtimeAPI) {
         // Editor
         if (!settings.disableEditor) {
             editor = require("./editor");
-            var editorApp = editor.init(server, settings, runtime, runtimeAPI);
+            var editorApp = editor.init(server, settings, runtimeAPI);
             adminApp.use(editorApp);
         }
 
diff --git a/red/red.js b/red/red.js
index 405364a94..717d74507 100644
--- a/red/red.js
+++ b/red/red.js
@@ -70,9 +70,8 @@ module.exports = {
         redUtil.init(userSettings);
         if (userSettings.httpAdminRoot !== false) {
             runtime.init(userSettings,redUtil,api);
-
             runtimeAPI.init(runtime,redUtil);
-            api.init(httpServer,userSettings,runtime,runtimeAPI,redUtil);
+            api.init(httpServer,userSettings,runtime.storage,runtimeAPI,redUtil);
 
             apiEnabled = true;
             server = runtime.adminApi.server;
diff --git a/red/runtime-api/comms.js b/red/runtime-api/comms.js
index 090a767bf..d89123795 100644
--- a/red/runtime-api/comms.js
+++ b/red/runtime-api/comms.js
@@ -15,5 +15,115 @@
  **/
 
 /**
- * @module red/comms
+ * @namespace RED.comms
  */
+
+/**
+ * @typedef CommsConnection
+ * @type {object}
+ * @property {string} session - a unique session identifier
+ * @property {Object} user - the user associated with the connection
+ * @property {Function} send - publish a message to the connection
+ */
+
+
+var runtime;
+var retained = {};
+var connections = [];
+
+
+function handleCommsEvent(event) {
+    publish(event.topic,event.data,event.retain);
+}
+function handleStatusEvent(event) {
+    publish("status/"+event.id,event.status,true);
+}
+function handleRuntimeEvent(event) {
+    runtime.log.trace("runtime event: "+JSON.stringify(event));
+    publish("notification/"+event.id,event.payload||{},event.retain);
+}
+
+function publish(topic,data,retain) {
+    if (retain) {
+        retained[topic] = data;
+    } else {
+        delete retained[topic];
+    }
+    connections.forEach(connection => connection.send(topic,data,retain))
+}
+
+
+var api = module.exports = {
+    init: function(_runtime) {
+        runtime = _runtime;
+        runtime.events.removeListener("node-status",handleStatusEvent);
+        runtime.events.on("node-status",handleStatusEvent);
+        runtime.events.removeListener("runtime-event",handleRuntimeEvent);
+        runtime.events.on("runtime-event",handleRuntimeEvent);
+        runtime.events.removeListener("comms",handleCommsEvent);
+        runtime.events.on("comms",handleCommsEvent);
+    },
+
+    /**
+    * Registers a new comms connection
+    * @param {Object} opts
+    * @param {User} opts.user - the user calling the api
+    * @param {CommsConnection} opts.client - the client connection
+    * @return {Promise<Object>} - resolves when complete
+    * @memberof RED.comms
+    */
+    addConnection: function(opts) {
+        connections.push(opts.client);
+        return Promise.resolve();
+    },
+
+    /**
+    * Unregisters a comms connection
+    * @param {Object} opts
+    * @param {User} opts.user - the user calling the api
+    * @param {CommsConnection} opts.client - the client connection
+    * @return {Promise<Object>} - resolves when complete
+    * @memberof RED.comms
+    */
+    removeConnection: function(opts) {
+        for (var i=0;i<connections.length;i++) {
+            if (connections[i] === opts.client) {
+                connections.splice(i,1);
+                break;
+            }
+        }
+        return Promise.resolve();
+    },
+
+    /**
+    * Subscribes a comms connection to a given topic. Currently, all clients get
+    * automatically subscribed to everything and cannot unsubscribe. Sending a subscribe
+    * request will trigger retained messages to be sent.
+    * @param {Object} opts
+    * @param {User} opts.user - the user calling the api
+    * @param {CommsConnection} opts.client - the client connection
+    * @param {String} opts.topic - the topic to subscribe to
+    * @return {Promise<Object>} - resolves when complete
+    * @memberof RED.comms
+    */
+    subscribe: function(opts) {
+        var re = new RegExp("^"+opts.topic.replace(/([\[\]\?\(\)\\\\$\^\*\.|])/g,"\\$1").replace(/\+/g,"[^/]+").replace(/\/#$/,"(\/.*)?")+"$");
+        for (var t in retained) {
+            if (re.test(t)) {
+                opts.client.send(t,retained[t]);
+            }
+        }
+        return Promise.resolve();
+    },
+
+    /**
+    * TODO: Unsubscribes a comms connection from a given topic
+    * @param {Object} opts
+    * @param {User} opts.user - the user calling the api
+    * @param {CommsConnection} opts.client - the client connection
+    * @param {String} opts.topic - the topic to unsubscribe from
+    * @return {Promise<Object>} - resolves when complete
+    * @memberof RED.comms
+    */
+    unsubscribe: function(opts) {}
+};
diff --git a/red/runtime-api/flows.js b/red/runtime-api/flows.js
index f572e6e7c..ba52090a0 100644
--- a/red/runtime-api/flows.js
+++ b/red/runtime-api/flows.js
@@ -20,7 +20,6 @@
 
 /**
  * @typedef Flows
- * @alias Dave
  * @type {object}
  * @property {string} rev - the flow revision identifier
  * @property {Array}  flows - the flow configuration, an array of node configuration objects
diff --git a/red/runtime-api/index.js b/red/runtime-api/index.js
index bdba811a6..3f12890ee 100644
--- a/red/runtime-api/index.js
+++ b/red/runtime-api/index.js
@@ -14,12 +14,11 @@
  * limitations under the License.
  **/
 
-
- /**
-  * A user accessing the API
-  * @typedef User
-  * @type {object}
-  */
+/**
+ * A user accessing the API
+ * @typedef User
+ * @type {object}
+ */
 
 var runtime;
 /**
@@ -28,6 +27,7 @@ var runtime;
 var api = module.exports = {
     init: function(_runtime, redUtil) {
         runtime = _runtime;
+        api.comms.init(runtime);
         api.flows.init(runtime);
         api.nodes.init(runtime);
         api.settings.init(runtime);
@@ -42,7 +42,24 @@ var api = module.exports = {
     settings: require("./settings"),
     projects: require("./projects"),
 
+    /**
+    * Returns whether the runtime is started
+    * @param {Object} opts
+    * @param {User} opts.user - the user calling the api
+    * @return {Promise<Boolean>} - whether the runtime is started
+    * @memberof RED
+    */
+    isStarted: function(opts) {
+        return Promise.resolve(runtime.isStarted());
+    },
 
+    /**
+    * Returns version number of the runtime
+    * @param {Object} opts
+    * @param {User} opts.user - the user calling the api
+    * @return {Promise<String>} - the runtime version number
+    * @memberof RED
+    */
     version: function(opts) {
         return Promise.resolve(runtime.version());
     }
diff --git a/red/runtime/nodes/registry/loader.js b/red/runtime/nodes/registry/loader.js
index 7f677f3da..c580939c3 100644
--- a/red/runtime/nodes/registry/loader.js
+++ b/red/runtime/nodes/registry/loader.js
@@ -71,6 +71,15 @@ function createNodeApi(node) {
         events: runtime.events,
         util: runtime.util,
         version: runtime.version,
+        comms: {
+            publish: function(topic,data,retain) {
+                runtime.events.emit("comms",{
+                    topic: topic,
+                    data: data,
+                    retain: retain
+                })
+            }
+        }
     }
     copyObjectProperties(runtime.nodes,red.nodes,["createNode","getNode","eachNode","addCredentials","getCredentials","deleteCredentials" ]);
     red.nodes.registerType = function(type,constructor,opts) {
@@ -79,7 +88,6 @@ function createNodeApi(node) {
     copyObjectProperties(runtime.log,red.log,null,["init"]);
     copyObjectProperties(runtime.settings,red.settings,null,["init","load","reset"]);
     if (runtime.adminApi) {
-        red.comms = runtime.adminApi.comms;
         red.library = {
             register: function(type) {
                 return runtime.library.registerType(node.id,type);