1
0
mirror of https://github.com/node-red/node-red.git synced 2023-10-10 13:36:53 +02:00
node-red/red/nodes.js
Nicholas O'Leary 0507578c98 Always clone messages to ensure no cross-pollution
Part of #85

A function that returns the same message to multiple outputs, where each output is wired to at most one node was not having its messages cloned due to the change I made in #85 to be slightly more efficient. By returning the same message to each output, cross pollution was possible.
2013-11-29 19:56:46 +00:00

373 lines
12 KiB
JavaScript

/**
* Copyright 2013 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 EventEmitter = require("events").EventEmitter;
var fs = require("fs");
var path = require("path");
var clone = require("clone");
var events = require("./events");
var storage = null;
function getCallerFilename(type) {
//if (type == "summary") {
// var err = new Error();
// console.log(err.stack);
// return null;
//}
// Save original Error.prepareStackTrace
var origPrepareStackTrace = Error.prepareStackTrace;
// Override with function that just returns `stack`
Error.prepareStackTrace = function (_, stack) {
return stack;
}
// Create a new `Error`, which automatically gets `stack`
var err = new Error();
// Evaluate `err.stack`, which calls our new `Error.prepareStackTrace`
var stack = err.stack;
// Restore original `Error.prepareStackTrace`
Error.prepareStackTrace = origPrepareStackTrace;
// Remove superfluous function call on stack
stack.shift();
stack.shift();
return stack[0].getFileName();
}
var registry = (function() {
var nodes = {};
var logHandlers = [];
var obj = {
add: function(n) {
nodes[n.id] = n;
n.on("log",function(msg) {
for (var i in logHandlers) {
logHandlers[i].emit("log",msg);
}
});
},
get: function(i) {
return nodes[i];
},
clear: function() {
events.emit("nodes-stopping");
for (var n in nodes) {
nodes[n].close();
}
events.emit("nodes-stopped");
nodes = {};
},
each: function(cb) {
for (var n in nodes) {
cb(nodes[n]);
}
},
addLogHandler: function(handler) {
logHandlers.push(handler);
}
}
return obj;
})();
var ConsoleLogHandler = new EventEmitter();
ConsoleLogHandler.on("log",function(msg) {
util.log("["+msg.level+"] ["+msg.type+":"+(msg.name||msg.id)+"] "+msg.msg);
});
registry.addLogHandler(ConsoleLogHandler);
var node_type_registry = (function() {
var node_types = {};
var node_configs = {};
var obj = {
register: function(type,node) {
util.inherits(node, Node);
var callerFilename = getCallerFilename(type);
if (callerFilename == null) {
util.log("["+type+"] unable to determine filename");
} else {
var configFilename = callerFilename.replace(/\.js$/,".html");
if (fs.existsSync(configFilename)) {
node_types[type] = node;
if (! node_configs[configFilename]) {
node_configs[configFilename] = fs.readFileSync(configFilename,'utf8');
}
events.emit("type-registered",type);
} else {
util.log("["+type+"] missing template file: "+configFilename);
}
}
},
get: function(type) {
return node_types[type];
},
getNodeConfigs: function() {
var result = "";
for (var nt in node_configs) {
result += node_configs[nt];
}
return result;
}
}
return obj;
})();
function Node(n) {
this.id = n.id;
registry.add(this);
this.type = n.type;
if (n.name) {
this.name = n.name;
}
this.wires = n.wires||[];
}
util.inherits(Node,EventEmitter);
Node.prototype.close = function() {
// called when a node is removed
this.emit("close");
}
Node.prototype.send = function(msg) {
// instanceof doesn't work for some reason here
if (msg == null) {
msg = [];
} else if (!util.isArray(msg)) {
msg = [msg];
}
for (var i in this.wires) {
var wires = this.wires[i];
if (i < msg.length) {
if (msg[i] != null) {
var msgs = msg[i];
if (!util.isArray(msg[i])) {
msgs = [msg[i]];
}
//if (wires.length == 1) {
// // Single recipient, don't need to clone the message
// var node = registry.get(wires[0]);
// if (node) {
// for (var k in msgs) {
// var mm = msgs[k];
// node.receive(mm);
// }
// }
//} else {
// Multiple recipients, must send message copies
for (var j in wires) {
var node = registry.get(wires[j]);
if (node) {
for (var k in msgs) {
var mm = msgs[k];
// Temporary fix for #97
// TODO: remove this http-node-specific fix somehow
var req = mm.req;
var res = mm.res;
mm.req = null;
mm.res = null;
var m = clone(mm);
m.req = req;
m.res = res;
node.receive(m);
}
}
}
//}
}
}
}
}
module.exports.Node = Node;
Node.prototype.receive = function(msg) {
this.emit("input",msg);
}
Node.prototype.log = function(msg) {
var o = {level:'log',id:this.id, type:this.type, msg:msg};
if (this.name) o.name = this.name;
this.emit("log",o);
}
Node.prototype.warn = function(msg) {
var o = {level:'warn',id:this.id, type:this.type, msg:msg};
if (this.name) o.name = this.name;
this.emit("log",o);
}
Node.prototype.error = function(msg) {
var o = {level:'error',id:this.id, type:this.type, msg:msg};
if (this.name) o.name = this.name;
this.emit("log",o);
}
var credentials = {};
module.exports.addCredentials = function(id,creds) {
credentials[id] = creds;
if (!storage) {
// Do this lazily to ensure the storage provider as been initialised
storage = require("./storage");
}
storage.saveCredentials(credentials);
}
module.exports.getCredentials = function(id) {
return credentials[id];
}
module.exports.deleteCredentials = function(id) {
delete credentials[id];
storage.saveCredentials(credentials);
}
module.exports.createNode = function(node,def) {
Node.call(node,def);
}
module.exports.registerType = node_type_registry.register;
module.exports.getNodeConfigs = node_type_registry.getNodeConfigs;
module.exports.addLogHandler = registry.addLogHandler;
module.exports.load = function(settings) {
function loadNodes(dir) {
fs.readdirSync(dir).sort().filter(function(fn){
var stats = fs.statSync(path.join(dir,fn));
if (stats.isFile()) {
if (/\.js$/.test(fn)) {
try {
require(path.join(dir,fn));
} catch(err) {
util.log("["+fn+"] "+err);
//console.log(err.stack);
}
}
} else if (stats.isDirectory()) {
// Ignore /.dirs/, /lib/ /node_modules/
if (!/^(\..*|lib|icons|node_modules)$/.test(fn)) {
loadNodes(path.join(dir,fn));
} else if (fn === "icons") {
events.emit("node-icon-dir",path.join(dir,fn));
}
}
});
}
loadNodes(__dirname+"/../nodes");
if (settings.nodesDir) {
loadNodes(settings.nodesDir);
}
//events.emit("nodes-loaded");
}
var activeConfig = [];
var missingTypes = [];
events.on('type-registered',function(type) {
if (missingTypes.length > 0) {
var i = missingTypes.indexOf(type);
if (i != -1) {
missingTypes.splice(i,1);
util.log("[red] Missing type registered: "+type);
}
if (missingTypes.length == 0) {
parseConfig();
}
}
});
module.exports.getNode = function(nid) {
return registry.get(nid);
}
function stopFlows() {
if (activeConfig&&activeConfig.length > 0) {
util.log("[red] Stopping flows");
}
registry.clear();
}
module.exports.stopFlows = stopFlows;
module.exports.setConfig = function(conf) {
stopFlows();
activeConfig = conf;
if (!storage) {
// Do this lazily to ensure the storage provider as been initialised
storage = require("./storage");
}
storage.getCredentials().then(function(creds) {
credentials = creds;
parseConfig();
}).otherwise(function(err) {
util.log("[red] Error loading credentials : "+err);
});
}
module.exports.getConfig = function() {
return activeConfig;
}
var parseConfig = function() {
missingTypes = [];
for (var i in activeConfig) {
var type = activeConfig[i].type;
// TODO: remove workspace in next release+1
if (type != "workspace" && type != "tab") {
var nt = node_type_registry.get(type);
if (!nt && missingTypes.indexOf(type) == -1) {
missingTypes.push(type);
}
}
};
if (missingTypes.length > 0) {
util.log("[red] Waiting for missing types to be registered:");
for (var i in missingTypes) {
util.log("[red] - "+missingTypes[i]);
}
return;
}
util.log("[red] Starting flows");
events.emit("nodes-starting");
for (var i in activeConfig) {
var nn = null;
// TODO: remove workspace in next release+1
if (activeConfig[i].type != "workspace" && activeConfig[i].type != "tab") {
var nt = node_type_registry.get(activeConfig[i].type);
if (nt) {
try {
nn = new nt(activeConfig[i]);
}
catch (err) {
util.log("[red] "+activeConfig[i].type+" : "+err);
}
}
// console.log(nn);
if (nn == null) {
util.log("[red] unknown type: "+activeConfig[i].type);
}
}
}
// Clean up any orphaned credentials
var deletedCredentials = false;
for (var c in credentials) {
var n = registry.get(c);
if (!n) {
deletedCredentials = true;
delete credentials[c];
}
}
if (deletedCredentials) {
storage.saveCredentials(credentials);
}
events.emit("nodes-started");
}