frodovdr/node-red
1
0
Fork 0
mirror of https://github.com/node-red/node-red.git synced 2025-03-01 10:36:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move comms from runtime to api component

Browse Source
This commit is contained in:
Nick O'Leary
2015-11-12 09:03:03 +00:00
parent 9f5e6a4b37
commit 8fb955e182
11 changed files with 145 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

207
red/runtime/comms.js
View File

@@ -1,207 +0,0 @@
/**
* Copyright 2014, 2015 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 ws = require("ws");
var log = require("./log");
var server;
var settings;
var wsServer;
var pendingConnections = [];
var activeConnections = [];
var retained = {};
var heartbeatTimer;
var lastSentTime;
function init(_server,_settings) {
server = _server;
settings = _settings;
}
function start() {
var Tokens = require("../api/auth/tokens");
var Users = require("../api/auth/users");
var Permissions = require("../api/auth/permissions");
if (!settings.disableEditor) {
Users.default().then(function(anonymousUser) {
var webSocketKeepAliveTime = settings.webSocketKeepAliveTime || 15000;
var path = settings.httpAdminRoot || "/";
path = (path.slice(0,1) != "/" ? "/":"") + path + (path.slice(-1) == "/" ? "":"/") + "comms";
wsServer = new ws.Server({server:server,path:path});
wsServer.on('connection',function(ws) {
log.audit({event: "comms.open"});
var pendingAuth = (settings.adminAuth != null);
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) {
if (!userScope || !Permissions.hasPermission(userScope,"status.read")) {
ws.close();
} else {
pendingAuth = false;
removePendingConnection(ws);
activeConnections.push(ws);
if (sendAck) {
ws.send(JSON.stringify({auth:"ok"}));
}
}
}
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);
} else {
log.audit({event: "comms.auth.fail"});
completeConnection(null,false);
}
//TODO: duplicated code - pull non-auth message handling out
if (msg.subscribe) {
handleRemoteSubscription(ws,msg.subscribe);
}
}
}
});
ws.on('error', function(err) {
log.warn(log._("comms.error",{message:err.toString()}));
});
});
wsServer.on('error', function(err) {
log.warn(log._("comms.error-server",{message:err.toString()}));
});
lastSentTime = Date.now();
heartbeatTimer = setInterval(function() {
var now = Date.now();
if (now-lastSentTime > webSocketKeepAliveTime) {
publish("hb",lastSentTime);
}
}, webSocketKeepAliveTime);
});
}
}
function stop() {
if (heartbeatTimer) {
clearInterval(heartbeatTimer);
heartbeatTimer = null;
}
if (wsServer) {
wsServer.close();
wsServer = null;
}
}
function publish(topic,data,retain) {
if (retain) {
retained[topic] = data;
} else {
delete retained[topic];
}
lastSentTime = Date.now();
activeConnections.forEach(function(conn) {
publishTo(conn,topic,data);
});
}
function publishTo(ws,topic,data) {
var msg = JSON.stringify({topic:topic,data:data});
try {
ws.send(msg);
} catch(err) {
removeActiveConnection(ws);
removePendingConnection(ws);
log.warn(log._("comms.error-send",{message:err.toString()}));
}
}
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) {
for (var i=0;i<activeConnections.length;i++) {
if (activeConnections[i] === ws) {
activeConnections.splice(i,1);
break;
}
}
}
function removePendingConnection(ws) {
for (var i=0;i<pendingConnections.length;i++) {
if (pendingConnections[i] === ws) {
pendingConnections.splice(i,1);
break;
}
}
}
module.exports = {
init:init,
start:start,
stop:stop,
publish:publish
}

9
red/runtime/index.js
View File

@@ -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")

2
red/runtime/nodes/Node.js
View File

@@ -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;

4
red/runtime/nodes/flows/index.js
View File

@@ -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 {