Add message router component

nodes/core/core/20-inject.js

@@ -67,21 +67,20 @@ module.exports = function(RED) {
                 this.error(err,msg);
             }
         });
+        this.on('close', function() {
+            if (node.interval_id != null) {
+                clearInterval(node.interval_id);
+                if (RED.settings.verbose) { node.log(RED._("inject.stopped")); }
+            } else if (node.cronjob != null) {
+                node.cronjob.stop();
+                if (RED.settings.verbose) { node.log(RED._("inject.stopped")); }
+                delete node.cronjob;
+            }
+        });
     }
 
     RED.nodes.registerType("inject",InjectNode);
 
-    InjectNode.prototype.close = function() {
-        if (this.interval_id != null) {
-            clearInterval(this.interval_id);
-            if (RED.settings.verbose) { this.log(RED._("inject.stopped")); }
-        } else if (this.cronjob != null) {
-            this.cronjob.stop();
-            if (RED.settings.verbose) { this.log(RED._("inject.stopped")); }
-            delete this.cronjob;
-        }
-    }
-
     RED.httpAdmin.post("/inject/:id", RED.auth.needsPermission("inject.write"), function(req,res) {
         var node = RED.nodes.getNode(req.params.id);
         if (node != null) {

red/runtime/nodes/Node.js

@@ -22,6 +22,7 @@ var redUtil = require("../util");
 var Log = require("../log");
 var context = require("./context");
 var flows = require("./flows");
+var router = require("./router");
 
 function Node(n) {
     this.id = n.id;
@@ -41,28 +42,10 @@ function Node(n) {
 util.inherits(Node, EventEmitter);
 
 Node.prototype.updateWires = function(wires) {
-    //console.log("UPDATE",this.id);
+    router.add(this,wires);
     this.wires = wires || [];
-    delete this._wire;
-
-    var wc = 0;
-    this.wires.forEach(function(w) {
-        wc+=w.length;
-    });
-    this._wireCount = wc;
-    if (wc === 0) {
-        // With nothing wired to the node, no-op send
-        this.send = function(msg) {}
-    } else {
-        this.send = Node.prototype.send;
-        if (this.wires.length === 1 && this.wires[0].length === 1) {
-            // Single wire, so we can shortcut the send when
-            // a single message is sent
-            this._wire = this.wires[0][0];
-        }
-    }
-
 }
+
 Node.prototype.context = function() {
     if (!this._context) {
          this._context = context.get(this._alias||this.id,this.z);
@@ -100,11 +83,13 @@ Node.prototype.close = function() {
     }
     if (promises.length > 0) {
         return when.settle(promises).then(function() {
+            router.remove(this);
             if (this._context) {
                 context.delete(this._alias||this.id,this.z);
             }
         });
     } else {
+        router.remove(this);
         if (this._context) {
             context.delete(this._alias||this.id,this.z);
         }
@@ -113,90 +98,7 @@ Node.prototype.close = function() {
 };
 
 Node.prototype.send = function(msg) {
-    var msgSent = false;
+    router.send(this,msg);
-    var node;
-
-    if (msg === null || typeof msg === "undefined") {
-        return;
-    } else if (!util.isArray(msg)) {
-        if (this._wire) {
-            // A single message and a single wire on output 0
-            // TODO: pre-load flows.get calls - cannot do in constructor
-            //       as not all nodes are defined at that point
-            if (!msg._msgid) {
-                msg._msgid = redUtil.generateId();
-            }
-            this.metric("send",msg);
-            node = flows.get(this._wire);
-            /* istanbul ignore else */
-            if (node) {
-                node.receive(msg);
-            }
-            return;
-        } else {
-            msg = [msg];
-        }
-    }
-
-    var numOutputs = this.wires.length;
-
-    // Build a list of send events so that all cloning is done before
-    // any calls to node.receive
-    var sendEvents = [];
-
-    var sentMessageId = null;
-
-    // for each output of node eg. [msgs to output 0, msgs to output 1, ...]
-    for (var i = 0; i < numOutputs; i++) {
-        var wires = this.wires[i]; // wires leaving output i
-        /* istanbul ignore else */
-        if (i < msg.length) {
-            var msgs = msg[i]; // msgs going to output i
-            if (msgs !== null && typeof msgs !== "undefined") {
-                if (!util.isArray(msgs)) {
-                    msgs = [msgs];
-                }
-                var k = 0;
-                // for each recipent node of that output
-                for (var j = 0; j < wires.length; j++) {
-                    node = flows.get(wires[j]); // node at end of wire j
-                    if (node) {
-                        // for each msg to send eg. [[m1, m2, ...], ...]
-                        for (k = 0; k < msgs.length; k++) {
-                            var m = msgs[k];
-                            if (m !== null && m !== undefined) {
-                                /* istanbul ignore else */
-                                if (!sentMessageId) {
-                                    sentMessageId = m._msgid;
-                                }
-                                if (msgSent) {
-                                    var clonedmsg = redUtil.cloneMessage(m);
-                                    sendEvents.push({n:node,m:clonedmsg});
-                                } else {
-                                    sendEvents.push({n:node,m:m});
-                                    msgSent = true;
-                                }
-                            }
-                        }
-                    }
-                }
-            }
-        }
-    }
-    /* istanbul ignore else */
-    if (!sentMessageId) {
-        sentMessageId = redUtil.generateId();
-    }
-    this.metric("send",{_msgid:sentMessageId});
-
-    for (i=0;i<sendEvents.length;i++) {
-        var ev = sendEvents[i];
-        /* istanbul ignore else */
-        if (!ev.m._msgid) {
-            ev.m._msgid = sentMessageId;
-        }
-        ev.n.receive(ev.m);
-    }
 };
 
 Node.prototype.receive = function(msg) {

red/runtime/nodes/index.js

@@ -14,10 +14,6 @@
  * limitations under the License.
  **/
 
-var when = require("when");
-var path = require("path");
-var fs = require("fs");
-
 var registry = require("./registry");
 var credentials = require("./credentials");
 var flows = require("./flows");

red/runtime/nodes/router/index.js

@@ -0,0 +1,112 @@
+/**
+ * Copyright 2016 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ **/
+
+var util = require("util");
+
+var flows = require("../flows");
+var redUtil = require("../../util");
+
+var runtime;
+
+var routes = {};
+var sendQueue = [];
+
+
+
+function init(_runtime) {
+    runtime = _runtime;
+    wires = {};
+}
+
+function add(sourceNode, wires) {
+    routes[sourceNode.id] = wires;
+}
+function remove(sourceNode) {
+    delete routes[sourceNode.id];
+}
+function send(sourceNode, msg) {
+    if (msg === null || typeof msg === "undefined") {
+        return;
+    } else if (!util.isArray(msg)) {
+        msg = [msg];
+    }
+    var node;
+    var msgSent = false;
+    var nodeWires = routes[sourceNode.id];
+    if (nodeWires) {
+        var numOutputs = nodeWires.length;
+        var sendEvents = [];
+        var sentMessageId = null;
+        // for each output of node eg. [msgs to output 0, msgs to output 1, ...]
+        for (var i = 0; i < numOutputs; i++) {
+            var wires = nodeWires[i];
+            /* istanbul ignore else */
+            if (i < msg.length) {
+                var msgs = msg[i]; // msgs going to output i
+                if (msgs !== null && typeof msgs !== "undefined") {
+                    if (!util.isArray(msgs)) {
+                        msgs = [msgs];
+                    }
+                    var k = 0;
+                    // for each recipent node of that output
+                    for (var j = 0; j < wires.length; j++) {
+                        node = flows.get(wires[j]); // node at end of wire j
+                        if (node) {
+                            // for each msg to send eg. [[m1, m2, ...], ...]
+                            for (k = 0; k < msgs.length; k++) {
+                                var m = msgs[k];
+                                if (m !== null && m !== undefined) {
+                                    /* istanbul ignore else */
+                                    if (!sentMessageId) {
+                                        sentMessageId = m._msgid;
+                                    }
+                                    if (msgSent) {
+                                        var clonedmsg = redUtil.cloneMessage(m);
+                                        sendEvents.push({n:node,m:clonedmsg});
+                                    } else {
+                                        sendEvents.push({n:node,m:m});
+                                        msgSent = true;
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }
+        /* istanbul ignore else */
+        if (!sentMessageId) {
+            sentMessageId = redUtil.generateId();
+        }
+        sourceNode.metric("send",{_msgid:sentMessageId});
+
+        for (i=0;i<sendEvents.length;i++) {
+            var ev = sendEvents[i];
+            /* istanbul ignore else */
+            if (!ev.m._msgid) {
+                ev.m._msgid = sentMessageId;
+            }
+            ev.n.receive(ev.m);
+        }
+    }
+}
+
+module.exports = {
+    init:init,
+    add:add,
+    remove: remove,
+    send:send
+}

test/red/runtime/nodes/Node_spec.js

@@ -149,11 +149,18 @@ describe('Node', function() {
     });
 
     describe('#send', function() {
+        var flowGet;
+        afterEach(function() {
+            if (flowGet && flowGet.restore) {
+                flowGet.restore();
+                flowGet = null;
+            }
+        });
 
         it('emits a single message', function(done) {
             var n1 = new RedNode({id:'n1',type:'abc',wires:[['n2']]});
             var n2 = new RedNode({id:'n2',type:'abc'});
-            var flowGet = sinon.stub(flows,"get",function(id) {
+            flowGet = sinon.stub(flows,"get",function(id) {
                 return {'n1':n1,'n2':n2}[id];
             });
             var message = {payload:"hello world"};
@@ -172,7 +179,7 @@ describe('Node', function() {
         it('emits multiple messages on a single output', function(done) {
             var n1 = new RedNode({id:'n1',type:'abc',wires:[['n2']]});
             var n2 = new RedNode({id:'n2',type:'abc'});
-            var flowGet = sinon.stub(flows,"get",function(id) {
+            flowGet = sinon.stub(flows,"get",function(id) {
                 return {'n1':n1,'n2':n2}[id];
             });
 
@@ -206,7 +213,7 @@ describe('Node', function() {
             var n3 = new RedNode({id:'n3',type:'abc'});
             var n4 = new RedNode({id:'n4',type:'abc'});
             var n5 = new RedNode({id:'n5',type:'abc'});
-            var flowGet = sinon.stub(flows,"get",function(id) {
+            flowGet = sinon.stub(flows,"get",function(id) {
                 return {'n1':n1,'n2':n2,'n3':n3,'n4':n4,'n5':n5}[id];
             });
 
@@ -264,7 +271,7 @@ describe('Node', function() {
         it('emits no messages', function(done) {
             var n1 = new RedNode({id:'n1',type:'abc',wires:[['n2']]});
             var n2 = new RedNode({id:'n2',type:'abc'});
-            var flowGet = sinon.stub(flows,"get",function(id) {
+            flowGet = sinon.stub(flows,"get",function(id) {
                 return {'n1':n1,'n2':n2}[id];
             });
 
@@ -283,7 +290,7 @@ describe('Node', function() {
         it('emits messages ignoring non-existent nodes', function(done) {
             var n1 = new RedNode({id:'n1',type:'abc',wires:[['n9'],['n2']]});
             var n2 = new RedNode({id:'n2',type:'abc'});
-            var flowGet = sinon.stub(flows,"get",function(id) {
+            flowGet = sinon.stub(flows,"get",function(id) {
                 return {'n1':n1,'n2':n2}[id];
             });
 
@@ -307,7 +314,7 @@ describe('Node', function() {
             var n1 = new RedNode({id:'n1',type:'abc',wires:[[['n2'],['n3']]]});
             var n2 = new RedNode({id:'n2',type:'abc'});
             var n3 = new RedNode({id:'n3',type:'abc'});
-            var flowGet = sinon.stub(flows,"get",function(id) {
+            flowGet = sinon.stub(flows,"get",function(id) {
                 return {'n1':n1,'n2':n2,'n3':n3}[id];
             });
 
@@ -349,7 +356,7 @@ describe('Node', function() {
         });
 
          it("logs the uuid for all messages sent", function(done) {
-            var flowGet = sinon.stub(flows,"get",function(id) {
+            flowGet = sinon.stub(flows,"get",function(id) {
                 return {'n1':sender,'n2':receiver1,'n3':receiver2}[id];
             });
             var logHandler = {

test/red/runtime/nodes/router/index_spec.js

@@ -0,0 +1,79 @@
+/**
+ * Copyright 2016 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ **/
+
+var should = require("should");
+var sinon = require('sinon');
+var RedNode = require("../../../../../red/runtime/nodes/Node");
+var router = require("../../../../../red/runtime/nodes/router");
+var flows = require("../../../../../red/runtime/nodes/flows");
+
+
+describe('Router', function() {
+    var flowGet;
+    afterEach(function() {
+        if (flowGet && flowGet.restore) {
+            flowGet.restore();
+            flowGet = null;
+        }
+    })
+
+    describe('#add',function() {
+        it('adds a route for a node', function(done) {
+            var senderNode = {id:'123',metric:function(){}};
+            var receiver = sinon.stub();
+            flowGet = sinon.stub(flows,"get",function(id) {
+                if (id === '456') {
+                    return {receive:receiver};
+                }
+                return null;
+            });
+            router.send(senderNode,{});
+            flowGet.called.should.be.false();
+
+            router.add(senderNode,[['456']]);
+
+            router.send(senderNode,{});
+            flowGet.called.should.be.true();
+            receiver.called.should.be.true();
+
+            done();
+        });
+    })
+
+    describe('#remove',function() {
+        it('removes a route for a node', function(done) {
+            var senderNode = {id:'123',metric:function(){}};
+            var receiver = sinon.stub();
+            flowGet = sinon.stub(flows,"get",function(id) {
+                if (id === '456') {
+                    return {receive:receiver};
+                }
+                return null;
+            });
+            router.add(senderNode,[['456']]);
+            router.send(senderNode,{});
+            flowGet.called.should.be.true();
+            receiver.called.should.be.true();
+            flowGet.reset();
+            receiver.reset();
+
+            router.remove(senderNode);
+            router.send(senderNode,{});
+            flowGet.called.should.be.false();
+            done();
+        });
+    })
+})