mirror of
https://github.com/node-red/node-red.git
synced 2023-10-10 13:36:53 +02:00
Fixes #85 As well as adding deep-clone (via the new dependency on the 'clone' module), we no longer clone the message if there is a single recipient. This makes simple node-to-node flows more efficient. I've done some simple profiling using process.hrtime to time how long the Node.send function takes, and at best, this change is neutral to performance.
361 lines
11 KiB
JavaScript
361 lines
11 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];
|
|
var m = clone(mm);
|
|
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;
|
|
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");
|
|
}
|