1
0
mirror of https://github.com/node-red/node-red.git synced 2023-10-10 13:36:53 +02:00
node-red/red/api/comms.js

247 lines
8.7 KiB
JavaScript
Raw Normal View History

2014-05-07 21:47:25 +02:00
/**
* Copyright JS Foundation and other contributors, http://js.foundation
2014-05-07 21:47:25 +02:00
*
* 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;
2014-05-07 21:47:25 +02:00
var server;
var settings;
var wsServer;
2014-11-12 14:21:59 +01:00
var pendingConnections = [];
2014-05-07 21:47:25 +02:00
var activeConnections = [];
2014-05-08 15:15:54 +02:00
var retained = {};
2014-05-07 21:47:25 +02:00
var heartbeatTimer;
var lastSentTime;
function handleStatus(event) {
publish("status/"+event.id,event.status,true);
}
function handleRuntimeEvent(event) {
publish("notification/"+event.id,event.payload||{},event.retain);
}
function init(_server,runtime) {
2014-05-07 21:47:25 +02:00
server = _server;
settings = runtime.settings;
log = runtime.log;
runtime.events.removeListener("node-status",handleStatus);
runtime.events.on("node-status",handleStatus);
runtime.events.removeListener("runtime-event",handleRuntimeEvent);
runtime.events.on("runtime-event",handleRuntimeEvent);
2014-05-07 21:47:25 +02:00
}
2014-12-10 15:16:07 +01:00
function start() {
var Tokens = require("./auth/tokens");
var Users = require("./auth/users");
var Permissions = require("./auth/permissions");
2014-09-22 15:33:26 +02:00
if (!settings.disableEditor) {
Users.default().then(function(anonymousUser) {
2014-12-10 15:16:07 +01:00
var webSocketKeepAliveTime = settings.webSocketKeepAliveTime || 15000;
var path = settings.httpAdminRoot || "/";
2015-03-21 18:42:06 +01:00
path = (path.slice(0,1) != "/" ? "/":"") + path + (path.slice(-1) == "/" ? "":"/") + "comms";
wsServer = new ws.Server({
server:server,
path:path,
// Disable the deflate option due to this issue
// https://github.com/websockets/ws/pull/632
// that is fixed in the 1.x release of the ws module
// that we cannot currently pickup as it drops node 0.10 support
//perMessageDeflate: false
});
2014-12-10 15:16:07 +01:00
wsServer.on('connection',function(ws) {
log.audit({event: "comms.open"});
2014-12-10 15:16:07 +01:00
var pendingAuth = (settings.adminAuth != null);
2014-11-12 14:21:59 +01:00
if (!pendingAuth) {
2014-12-10 15:16:07 +01:00
activeConnections.push(ws);
2014-11-12 14:21:59 +01:00
} else {
pendingConnections.push(ws);
2014-09-22 15:33:26 +02:00
}
2014-12-10 15:16:07 +01:00
ws.on('close',function() {
log.audit({event: "comms.close",user:ws.user});
2014-12-10 15:16:07 +01:00
removeActiveConnection(ws);
removePendingConnection(ws);
});
ws.on('message', function(data,flags) {
var msg = null;
try {
msg = JSON.parse(data);
} catch(err) {
2015-05-08 15:21:01 +02:00
log.trace("comms received malformed message : "+err.toString());
2014-12-10 15:16:07 +01:00
return;
2014-11-12 14:21:59 +01:00
}
2014-12-10 15:16:07 +01:00
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.
2014-11-12 14:21:59 +01:00
}
2014-12-10 15:16:07 +01:00
}
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);
}
});
2014-12-10 15:16:07 +01:00
} else {
log.audit({event: "comms.auth.fail"});
completeConnection(null,false);
2014-12-10 15:16:07 +01:00
}
});
} 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);
}
2014-12-10 15:16:07 +01:00
}
2014-11-12 14:21:59 +01:00
}
2014-12-10 15:16:07 +01:00
});
ws.on('error', function(err) {
2015-05-08 15:21:01 +02:00
log.warn(log._("comms.error",{message:err.toString()}));
2014-12-10 15:16:07 +01:00
});
2014-09-22 15:33:26 +02:00
});
wsServer.on('error', function(err) {
2015-05-08 15:21:01 +02:00
log.warn(log._("comms.error-server",{message:err.toString()}));
2014-09-22 15:33:26 +02:00
});
2014-12-10 15:16:07 +01:00
lastSentTime = Date.now();
2014-12-10 15:16:07 +01:00
heartbeatTimer = setInterval(function() {
var now = Date.now();
if (now-lastSentTime > webSocketKeepAliveTime) {
publish("hb",lastSentTime);
}
2014-12-10 15:16:07 +01:00
}, webSocketKeepAliveTime);
2014-05-08 15:15:54 +02:00
});
2014-09-22 15:33:26 +02:00
}
2014-05-07 21:47:25 +02:00
}
function stop() {
2014-09-22 15:33:26 +02:00
if (heartbeatTimer) {
clearInterval(heartbeatTimer);
2015-03-21 18:42:06 +01:00
heartbeatTimer = null;
2014-09-22 15:33:26 +02:00
}
if (wsServer) {
wsServer.close();
2015-03-21 18:42:06 +01:00
wsServer = null;
2014-09-22 15:33:26 +02:00
}
}
2014-05-08 15:15:54 +02:00
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);
});
2014-05-08 15:15:54 +02:00
}
2014-05-07 21:47:25 +02:00
}
var stack = [];
var ok2tx = true;
2014-05-08 15:15:54 +02:00
function publishTo(ws,topic,data) {
if (topic && data) { stack.push({topic:topic,data:data}); }
if (ok2tx && (stack.length > 0)) {
ok2tx = false;
try {
ws.send(JSON.stringify(stack));
} catch(err) {
removeActiveConnection(ws);
removePendingConnection(ws);
log.warn(log._("comms.error-send",{message:err.toString()}));
}
stack = [];
var txtout = setTimeout(function() {
ok2tx = true;
publishTo(ws);
}, 50); // TODO: OK so a 50mS update rate should prob not be hard-coded
2014-05-08 15:15:54 +02:00
}
2014-05-08 15:15:54 +02:00
}
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]);
}
}
}
2014-11-12 14:21:59 +01:00
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;
}
}
}
2014-05-07 21:47:25 +02:00
module.exports = {
init:init,
start:start,
stop:stop,
publish:publish
2014-05-07 21:47:25 +02:00
}