From 8fb955e182d67dd30d792161008467a9b8024fa0 Mon Sep 17 00:00:00 2001 From: Nick O'Leary Date: Thu, 12 Nov 2015 09:03:03 +0000 Subject: [PATCH] Move comms from runtime to api component --- nodes/core/core/58-debug.js | 1 - red/{runtime => api}/comms.js | 37 ++++++---- red/api/index.js | 19 ++++- red/red.js | 25 +++++-- red/runtime/index.js | 9 +-- red/runtime/nodes/Node.js | 2 - red/runtime/nodes/flows/index.js | 4 ++ test/red/{runtime => api}/comms_spec.js | 92 ++++++++++++++++++------- test/red/api/index_spec.js | 9 +-- test/red/runtime/index_spec.js | 43 ++---------- test/red/runtime/nodes/Node_spec.js | 13 ---- 11 files changed, 145 insertions(+), 109 deletions(-) rename red/{runtime => api}/comms.js (90%) rename test/red/{runtime => api}/comms_spec.js (81%) diff --git a/nodes/core/core/58-debug.js b/nodes/core/core/58-debug.js index d7d13f067..e0bae3ca1 100644 --- a/nodes/core/core/58-debug.js +++ b/nodes/core/core/58-debug.js @@ -124,7 +124,6 @@ module.exports = function(RED) { if (msg.msg.length > debuglength) { msg.msg = msg.msg.substr(0,debuglength) +" ...."; } - RED.comms.publish("debug",msg); } diff --git a/red/runtime/comms.js b/red/api/comms.js similarity index 90% rename from red/runtime/comms.js rename to red/api/comms.js index 59c8d0e11..de8960c1c 100644 --- a/red/runtime/comms.js +++ b/red/api/comms.js @@ -15,7 +15,7 @@ **/ var ws = require("ws"); -var log = require("./log"); +var log; var server; var settings; @@ -29,17 +29,24 @@ var retained = {}; var heartbeatTimer; var lastSentTime; +function handleStatus(event) { + publish("status/"+event.id,event.status,true); +} -function init(_server,_settings) { +function init(_server,runtime) { server = _server; - settings = _settings; + settings = runtime.settings; + log = runtime.log; + + runtime.events.removeListener("node-status",handleStatus); + runtime.events.on("node-status",handleStatus); } function start() { - var Tokens = require("../api/auth/tokens"); - var Users = require("../api/auth/users"); - var Permissions = require("../api/auth/permissions"); + 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; @@ -151,15 +158,17 @@ function stop() { } function publish(topic,data,retain) { - if (retain) { - retained[topic] = data; - } else { - delete retained[topic]; + if (server) { + if (retain) { + retained[topic] = data; + } else { + delete retained[topic]; + } + lastSentTime = Date.now(); + activeConnections.forEach(function(conn) { + publishTo(conn,topic,data); + }); } - lastSentTime = Date.now(); - activeConnections.forEach(function(conn) { - publishTo(conn,topic,data); - }); } function publishTo(ws,topic,data) { diff --git a/red/api/index.js b/red/api/index.js index e2cc5f92a..dde167dc3 100644 --- a/red/api/index.js +++ b/red/api/index.js @@ -19,6 +19,7 @@ var bodyParser = require("body-parser"); var util = require('util'); var path = require('path'); var passport = require('passport'); +var when = require('when'); var ui = require("./ui"); var nodes = require("./nodes"); @@ -28,6 +29,7 @@ var info = require("./info"); var theme = require("./theme"); var locales = require("./locales"); var credentials = require("./credentials"); +var comms = require("./comms"); var auth = require("./auth"); var needsPermission = auth.needsPermission; @@ -46,13 +48,14 @@ var errorHandler = function(err,req,res,next) { res.status(400).json({error:"unexpected_error", message:err.toString()}); }; -function init(runtime) { +function init(server,runtime) { var settings = runtime.settings; log = runtime.log; if (settings.httpNodeRoot !== false) { nodeApp = express(); } if (settings.httpAdminRoot !== false) { + comms.init(server,runtime); adminApp = express(); auth.init(runtime); credentials.init(runtime); @@ -123,15 +126,27 @@ function init(runtime) { adminApp.use(errorHandler); } } - +function start() { + comms.start(); + return when.resolve(); +} +function stop() { + comms.stop(); + return when.resolve(); +} module.exports = { init: init, + start: start, + stop: stop, library: { register: library.register }, auth: { needsPermission: auth.needsPermission }, + comms: { + publish: comms.publish + }, adminApp: function() { return adminApp; }, nodeApp: function() { return nodeApp; } }; diff --git a/red/red.js b/red/red.js index aead539fa..52328ff19 100644 --- a/red/red.js +++ b/red/red.js @@ -25,6 +25,7 @@ process.env.NODE_RED_HOME = process.env.NODE_RED_HOME || path.resolve(__dirname+ var nodeApp = null; var adminApp = null; var server = null; +var apiEnabled = false; function checkBuild() { var editorFile = path.resolve(path.join(__dirname,"..","public","red","red.min.js")); @@ -44,9 +45,9 @@ var RED = { if (!userSettings.SKIP_BUILD_CHECK) { checkBuild(); } - runtime.init(httpServer,userSettings); + runtime.init(userSettings); if (userSettings.httpAdminRoot !== false || userSettings.httpNodeRoot !== false) { - api.init(runtime); + api.init(server,runtime); adminApp = api.adminApp(); nodeApp = api.nodeApp(); } @@ -58,15 +59,29 @@ var RED = { put: function(){}, delete: function(){} } + } else { + apiEnabled = true; } return runtime.app; }, - start: runtime.start, - stop: runtime.stop, + start: function() { + return runtime.start().then(function() { + if (apiEnabled) { + return api.start(); + } + }); + }, + stop: function() { + return runtime.stop().then(function() { + if (apiEnabled) { + return api.stop(); + } + }) + }, nodes: runtime.api, events: runtime.events, log: runtime.log, - comms: runtime.comms, + comms: api.comms, settings:runtime.settings, util: runtime.util, version: runtime.version, diff --git a/red/runtime/index.js b/red/runtime/index.js index 85935660b..438bda900 100644 --- a/red/runtime/index.js +++ b/red/runtime/index.js @@ -17,7 +17,6 @@ var when = require('when'); var redNodes = require("./nodes"); -var comms = require("./comms"); var storage = require("./storage"); var log = require("./log"); var i18n = require("./i18n"); @@ -28,11 +27,10 @@ var fs = require("fs"); var runtimeMetricInterval = null; -function init(server,userSettings) { +function init(userSettings) { userSettings.version = version(); log.init(userSettings); settings.init(userSettings); - comms.init(server,settings); } function version() { @@ -105,7 +103,6 @@ function start() { } log.info(log._("runtime.paths.settings",{path:settings.settingsFile})); redNodes.loadFlows().then(redNodes.startFlows); - comms.start(); }).otherwise(function(err) { console.log(err); }); @@ -137,8 +134,7 @@ function stop() { clearInterval(runtimeMetricInterval); runtimeMetricInterval = null; } - redNodes.stopFlows(); - comms.stop(); + return redNodes.stopFlows(); } var runtime = module.exports = { @@ -152,7 +148,6 @@ var runtime = module.exports = { i18n: i18n, settings: settings, storage: storage, - comms: comms, events: events, api: redNodes, util: require("./util") diff --git a/red/runtime/nodes/Node.js b/red/runtime/nodes/Node.js index 1bebaa0e1..f167ac96b 100644 --- a/red/runtime/nodes/Node.js +++ b/red/runtime/nodes/Node.js @@ -22,7 +22,6 @@ var redUtil = require("../util"); var Log = require("../log"); var flows = require("./flows"); -var comms = require("../comms"); function Node(n) { this.id = n.id; @@ -250,7 +249,6 @@ Node.prototype.metric = function(eventname, msg, metricValue) { * status: { fill:"red|green", shape:"dot|ring", text:"blah" } */ Node.prototype.status = function(status) { - comms.publish("status/" + this.id, status, true); flows.handleStatus(this,status); }; module.exports = Node; diff --git a/red/runtime/nodes/flows/index.js b/red/runtime/nodes/flows/index.js index bab9a68d6..5b251df46 100644 --- a/red/runtime/nodes/flows/index.js +++ b/red/runtime/nodes/flows/index.js @@ -182,6 +182,10 @@ function delegateStatus(node,statusMessage) { } } function handleStatus(node,statusMessage) { + events.emit("node-status",{ + id: node.id, + status:statusMessage + }); if (node.z) { delegateStatus(node,statusMessage); } else { diff --git a/test/red/runtime/comms_spec.js b/test/red/api/comms_spec.js similarity index 81% rename from test/red/runtime/comms_spec.js rename to test/red/api/comms_spec.js index 53aff1a41..2fab2016a 100644 --- a/test/red/runtime/comms_spec.js +++ b/test/red/api/comms_spec.js @@ -23,21 +23,26 @@ var express = require('express'); var app = express(); var WebSocket = require('ws'); -var comms = require("../../../red/runtime/comms"); +var comms = require("../../../red/api/comms"); var Users = require("../../../red/api/auth/users"); var Tokens = require("../../../red/api/auth/tokens"); var address = '127.0.0.1'; var listenPort = 0; // use ephemeral port -describe("runtime/comms", function() { +describe("api/comms", function() { describe("with default keepalive", function() { var server; var url; var port; before(function(done) { + sinon.stub(Users,"default",function() { return when.resolve(null);}); server = http.createServer(function(req,res){app(req,res)}); - comms.init(server, {}); + comms.init(server, { + settings:{}, + log:{warn:function(){},_:function(){},trace:function(){},audit:function(){}}, + events:{on:function(){},removeListener:function(){}} + }); server.listen(listenPort, address); server.on('listening', function() { port = server.address().port; @@ -48,6 +53,7 @@ describe("runtime/comms", function() { }); after(function() { + Users.default.restore(); comms.stop(); }); @@ -119,29 +125,34 @@ describe("runtime/comms", function() { // implementation. More test should be written to test topic // matching once this one is passing - if (0) { - it('receives message on correct topic', function(done) { - var ws = new WebSocket(url); - ws.on('open', function() { - ws.send('{"subscribe":"topic4"}'); - comms.publish('topic5', 'foo'); - comms.publish('topic4', 'bar'); - }); - ws.on('message', function(msg) { - msg.should.equal('{"topic":"topic4","data":"bar"}'); - ws.close(); - done(); - }); + it.skip('receives message on correct topic', function(done) { + var ws = new WebSocket(url); + ws.on('open', function() { + ws.send('{"subscribe":"topic4"}'); + comms.publish('topic5', 'foo'); + comms.publish('topic4', 'bar'); }); - } + ws.on('message', function(msg) { + msg.should.equal('{"topic":"topic4","data":"bar"}'); + ws.close(); + done(); + }); + }); + + it.skip('listens for node/status events'); }); describe("disabled editor", function() { var server; var url; var port; before(function(done) { + sinon.stub(Users,"default",function() { return when.resolve(null);}); server = http.createServer(function(req,res){app(req,res)}); - comms.init(server, {disableEditor:true}); + comms.init(server, { + settings:{disableEditor:true}, + log:{warn:function(){},_:function(){},trace:function(){},audit:function(){}}, + events:{on:function(){},removeListener:function(){}} + }); server.listen(listenPort, address); server.on('listening', function() { port = server.address().port; @@ -152,6 +163,7 @@ describe("runtime/comms", function() { }); after(function() { + Users.default.restore(); comms.stop(); }); @@ -173,8 +185,13 @@ describe("runtime/comms", function() { var url; var port; before(function(done) { + sinon.stub(Users,"default",function() { return when.resolve(null);}); server = http.createServer(function(req,res){app(req,res)}); - comms.init(server, {httpAdminRoot:"/adminPath"}); + comms.init(server, { + settings:{httpAdminRoot:"/adminPath"}, + log:{warn:function(){},_:function(){},trace:function(){},audit:function(){}}, + events:{on:function(){},removeListener:function(){}} + }); server.listen(listenPort, address); server.on('listening', function() { port = server.address().port; @@ -185,6 +202,7 @@ describe("runtime/comms", function() { }); after(function() { + Users.default.restore(); comms.stop(); }); @@ -206,8 +224,13 @@ describe("runtime/comms", function() { var url; var port; before(function(done) { + sinon.stub(Users,"default",function() { return when.resolve(null);}); server = http.createServer(function(req,res){app(req,res)}); - comms.init(server, {httpAdminRoot:"/adminPath/"}); + comms.init(server,{ + settings:{httpAdminRoot:"/adminPath"}, + log:{warn:function(){},_:function(){},trace:function(){},audit:function(){}}, + events:{on:function(){},removeListener:function(){}} + }); server.listen(listenPort, address); server.on('listening', function() { port = server.address().port; @@ -218,6 +241,7 @@ describe("runtime/comms", function() { }); after(function() { + Users.default.restore(); comms.stop(); }); @@ -239,8 +263,13 @@ describe("runtime/comms", function() { var url; var port; before(function(done) { + sinon.stub(Users,"default",function() { return when.resolve(null);}); server = http.createServer(function(req,res){app(req,res)}); - comms.init(server, {httpAdminRoot:"adminPath"}); + comms.init(server, { + settings:{httpAdminRoot:"adminPath"}, + log:{warn:function(){},_:function(){},trace:function(){},audit:function(){}}, + events:{on:function(){},removeListener:function(){}} + }); server.listen(listenPort, address); server.on('listening', function() { port = server.address().port; @@ -251,6 +280,7 @@ describe("runtime/comms", function() { }); after(function() { + Users.default.restore(); comms.stop(); }); @@ -272,8 +302,13 @@ describe("runtime/comms", function() { var url; var port; before(function(done) { + sinon.stub(Users,"default",function() { return when.resolve(null);}); server = http.createServer(function(req,res){app(req,res)}); - comms.init(server, {webSocketKeepAliveTime: 100}); + comms.init(server, { + settings:{webSocketKeepAliveTime: 100}, + log:{warn:function(){},_:function(){},trace:function(){},audit:function(){}}, + events:{on:function(){},removeListener:function(){}} + }); server.listen(listenPort, address); server.on('listening', function() { port = server.address().port; @@ -283,6 +318,7 @@ describe("runtime/comms", function() { }); }); after(function() { + Users.default.restore(); comms.stop(); }); it('are sent', function(done) { @@ -354,7 +390,11 @@ describe("runtime/comms", function() { server = http.createServer(function(req,res){app(req,res)}); - comms.init(server, {adminAuth:{}}); + comms.init(server,{ + settings:{adminAuth:{}}, + log:{warn:function(){},_:function(){},trace:function(){},audit:function(){}}, + events:{on:function(){},removeListener:function(){}} + }); server.listen(listenPort, address); server.on('listening', function() { port = server.address().port; @@ -440,7 +480,11 @@ describe("runtime/comms", function() { before(function(done) { getDefaultUser = sinon.stub(Users,"default",function() { return when.resolve({permissions:"read"});}); server = http.createServer(function(req,res){app(req,res)}); - comms.init(server, {adminAuth:{}}); + comms.init(server, { + settings:{adminAuth:{}}, + log:{warn:function(){},_:function(){},trace:function(){},audit:function(){}}, + events:{on:function(){},removeListener:function(){}} + }); server.listen(listenPort, address); server.on('listening', function() { port = server.address().port; diff --git a/test/red/api/index_spec.js b/test/red/api/index_spec.js index cf4ec02d6..e2f68ce00 100644 --- a/test/red/api/index_spec.js +++ b/test/red/api/index_spec.js @@ -28,7 +28,7 @@ describe("api index", function() { describe("disables editor", function() { before(function() { - api.init({ + api.init({},{ settings:{httpNodeRoot:true, httpAdminRoot: true,disableEditor:true}, events: {on:function(){},removeListener: function(){}} }); @@ -67,9 +67,10 @@ describe("api index", function() { }) }); before(function() { - api.init({ + api.init({},{ settings:{httpNodeRoot:true, httpAdminRoot: true, adminAuth:{type: "credentials",users:[],default:{permissions:"read"}}}, - storage:{getSessions:function(){return when.resolve({})}} + storage:{getSessions:function(){return when.resolve({})}}, + events:{on:function(){},removeListener:function(){}} }); app = api.adminApp(); }); @@ -103,7 +104,7 @@ describe("api index", function() { }); before(function() { - api.init({ + api.init({},{ log:{audit:function(){}}, settings:{httpNodeRoot:true, httpAdminRoot: true,disableEditor:false}, events:{on:function(){},removeListener:function(){}} diff --git a/test/red/runtime/index_spec.js b/test/red/runtime/index_spec.js index 17ad2ac14..50ff9b9e2 100644 --- a/test/red/runtime/index_spec.js +++ b/test/red/runtime/index_spec.js @@ -21,19 +21,12 @@ var path = require("path"); var api = require("../../../red/api"); var runtime = require("../../../red/runtime"); -var comms = require("../../../red/runtime/comms"); var redNodes = require("../../../red/runtime/nodes"); var storage = require("../../../red/runtime/storage"); var settings = require("../../../red/runtime/settings"); var log = require("../../../red/runtime/log"); describe("runtime", function() { - var commsMessages = []; - var commsPublish; - - beforeEach(function() { - commsMessages = []; - }); afterEach(function() { if (console.log.restore) { console.log.restore(); @@ -41,13 +34,9 @@ describe("runtime", function() { }) before(function() { - commsPublish = sinon.stub(comms,"publish", function(topic,msg,retained) { - commsMessages.push({topic:topic,msg:msg,retained:retained}); - }); process.env.NODE_RED_HOME = path.resolve(path.join(__dirname,"..","..","..")) }); after(function() { - commsPublish.restore(); delete process.env.NODE_RED_HOME; }); @@ -55,32 +44,26 @@ describe("runtime", function() { beforeEach(function() { sinon.stub(log,"init",function() {}); sinon.stub(settings,"init",function() {}); - sinon.stub(comms,"init",function() {}); }); afterEach(function() { log.init.restore(); settings.init.restore(); - comms.init.restore(); }) it("initialises components", function() { - var dummyServer = {}; - runtime.init(dummyServer,{testSettings: true, httpAdminRoot:"/"}); + runtime.init({testSettings: true, httpAdminRoot:"/"}); log.init.called.should.be.true; settings.init.called.should.be.true; - comms.init.called.should.be.true; }); it("returns version", function() { - var dummyServer = {}; - runtime.init(dummyServer,{testSettings: true, httpAdminRoot:"/"}); + runtime.init({testSettings: true, httpAdminRoot:"/"}); /^\d+\.\d+\.\d+(-git)?$/.test(runtime.version()).should.be.true; }) }); describe("start",function() { - var commsInit; var storageInit; var settingsLoad; var logMetric; @@ -93,10 +76,8 @@ describe("runtime", function() { var redNodesGetNodeList; var redNodesLoadFlows; var redNodesStartFlows; - var commsStart; beforeEach(function() { - commsInit = sinon.stub(comms,"init",function() {}); storageInit = sinon.stub(storage,"init",function(settings) {return when.resolve();}); logMetric = sinon.stub(log,"metric",function() { return false; }); logWarn = sinon.stub(log,"warn",function() { }); @@ -107,10 +88,8 @@ describe("runtime", function() { redNodesCleanModuleList = sinon.stub(redNodes,"cleanModuleList",function(){}); redNodesLoadFlows = sinon.stub(redNodes,"loadFlows",function() {return when.resolve()}); redNodesStartFlows = sinon.stub(redNodes,"startFlows",function() {}); - commsStart = sinon.stub(comms,"start",function(){}); }); afterEach(function() { - commsInit.restore(); storageInit.restore(); logMetric.restore(); logWarn.restore(); @@ -122,7 +101,6 @@ describe("runtime", function() { redNodesCleanModuleList.restore(); redNodesLoadFlows.restore(); redNodesStartFlows.restore(); - commsStart.restore(); }); it("reports errored/missing modules",function(done) { redNodesGetNodeList = sinon.stub(redNodes,"getNodeList", function(cb) { @@ -131,7 +109,7 @@ describe("runtime", function() { { module:"module",enabled:true,loaded:false,types:["typeA","typeB"]} // missing ].filter(cb); }); - runtime.init({},{testSettings: true, httpAdminRoot:"/", load:function() { return when.resolve();}}); + runtime.init({testSettings: true, httpAdminRoot:"/", load:function() { return when.resolve();}}); sinon.stub(console,"log"); runtime.start().then(function() { console.log.restore(); @@ -139,7 +117,6 @@ describe("runtime", function() { storageInit.calledOnce.should.be.true; redNodesInit.calledOnce.should.be.true; redNodesLoad.calledOnce.should.be.true; - commsStart.calledOnce.should.be.true; redNodesLoadFlows.calledOnce.should.be.true; logWarn.calledWithMatch("Failed to register 1 node type"); @@ -162,7 +139,7 @@ describe("runtime", function() { ].filter(cb); }); var serverInstallModule = sinon.stub(redNodes,"installModule",function(name) { return when.resolve();}); - runtime.init({},{testSettings: true, autoInstallModules:true, httpAdminRoot:"/", load:function() { return when.resolve();}}); + runtime.init({testSettings: true, autoInstallModules:true, httpAdminRoot:"/", load:function() { return when.resolve();}}); sinon.stub(console,"log"); runtime.start().then(function() { console.log.restore(); @@ -188,7 +165,7 @@ describe("runtime", function() { { err:"errored",name:"errName" } // error ].filter(cb); }); - runtime.init({},{testSettings: true, verbose:true, httpAdminRoot:"/", load:function() { return when.resolve();}}); + runtime.init({testSettings: true, verbose:true, httpAdminRoot:"/", load:function() { return when.resolve();}}); sinon.stub(console,"log"); runtime.start().then(function() { console.log.restore(); @@ -203,12 +180,11 @@ describe("runtime", function() { }); it("reports runtime metrics",function(done) { - var commsStop = sinon.stub(comms,"stop",function() {} ); var stopFlows = sinon.stub(redNodes,"stopFlows",function() {} ); redNodesGetNodeList = sinon.stub(redNodes,"getNodeList", function() {return []}); logMetric.restore(); logMetric = sinon.stub(log,"metric",function() { return true; }); - runtime.init({},{testSettings: true, runtimeMetricInterval:200, httpAdminRoot:"/", load:function() { return when.resolve();}}); + runtime.init({testSettings: true, runtimeMetricInterval:200, httpAdminRoot:"/", load:function() { return when.resolve();}}); sinon.stub(console,"log"); runtime.start().then(function() { console.log.restore(); @@ -226,7 +202,6 @@ describe("runtime", function() { done(err); } finally { runtime.stop(); - commsStop.restore(); stopFlows.restore(); } },300); @@ -237,15 +212,9 @@ describe("runtime", function() { }); it("stops components", function() { - var commsStop = sinon.stub(comms,"stop",function() {} ); var stopFlows = sinon.stub(redNodes,"stopFlows",function() {} ); - runtime.stop(); - - commsStop.called.should.be.true; stopFlows.called.should.be.true; - - commsStop.restore(); stopFlows.restore(); }); }); diff --git a/test/red/runtime/nodes/Node_spec.js b/test/red/runtime/nodes/Node_spec.js index 2cc35f9d5..12af38c97 100644 --- a/test/red/runtime/nodes/Node_spec.js +++ b/test/red/runtime/nodes/Node_spec.js @@ -19,7 +19,6 @@ var sinon = require('sinon'); var RedNode = require("../../../../red/runtime/nodes/Node"); var Log = require("../../../../red/runtime/log"); var flows = require("../../../../red/runtime/nodes/flows"); -var comms = require("../../../../red/runtime/comms"); describe('Node', function() { describe('#constructor',function() { @@ -512,9 +511,6 @@ describe('Node', function() { }); describe('#status', function() { - after(function() { - comms.publish.restore(); - }); it('publishes status', function(done) { sinon.stub(flows,"handleStatus", function(node,message,msg) {}); var n = new RedNode({id:'123',type:'abc'}); @@ -522,18 +518,9 @@ describe('Node', function() { var topic; var message; var retain; - sinon.stub(comms, 'publish', function(_topic, _message, _retain) { - topic = _topic; - message = _message; - retain = _retain; - }); n.status(status); - topic.should.equal('status/123'); - message.should.equal(status); - retain.should.be.true; - flows.handleStatus.called.should.be.true; flows.handleStatus.args[0][0].should.eql(n); flows.handleStatus.args[0][1].should.eql(status);