1
0
mirror of https://github.com/node-red/node-red.git synced 2023-10-10 13:36:53 +02:00
node-red/packages/node_modules/@node-red/runtime/lib/nodes/Node.js

677 lines
21 KiB
JavaScript
Raw Normal View History

2014-05-03 23:26:35 +02:00
/**
* Copyright JS Foundation and other contributors, http://js.foundation
2014-05-03 23:26:35 +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.
**/
2014-05-03 23:26:35 +02:00
var util = require("util");
var EventEmitter = require("events").EventEmitter;
2018-08-17 23:10:54 +02:00
var redUtil = require("@node-red/util").util;
2019-08-06 15:27:56 +02:00
var Log = require("@node-red/util").log;
2015-12-29 23:16:51 +01:00
var context = require("./context");
var flows = require("../flows");
2021-04-12 21:30:31 +02:00
const hooks = require("@node-red/util").hooks;
2014-05-03 23:26:35 +02:00
2019-08-06 15:27:56 +02:00
const NOOP_SEND = function() {}
/**
* The Node object is the heart of a Node-RED flow. It is the object that all
* nodes extend.
*
* The Node object itself inherits from EventEmitter, although it provides
* custom implementations of some of the EE functions in order to handle
* `input` and `close` events properly.
*/
2014-05-03 23:26:35 +02:00
function Node(n) {
this.id = n.id;
this.type = n.type;
2015-02-20 02:17:24 +01:00
this.z = n.z;
this.g = n.g;
this._closeCallbacks = [];
2019-07-09 00:23:33 +02:00
this._inputCallback = null;
this._inputCallbacks = null;
2014-05-03 23:26:35 +02:00
if (n.name) {
this.name = n.name;
}
if (n._alias) {
this._alias = n._alias;
}
2019-01-16 17:27:19 +01:00
if (n._flow) {
// Make this a non-enumerable property as it may cause
// circular references. Any existing code that tries to JSON serialise
// the object (such as dashboard) will not like circular refs
// The value must still be writable in the case that a node does:
// Object.assign(this,config)
// as part of its constructor - config._flow will overwrite this._flow
// which we can tolerate as they are the same object.
Object.defineProperty(this,'_flow', {value: n._flow, enumerable: false, writable: true })
2019-01-16 17:27:19 +01:00
}
if (n._module) {
Object.defineProperty(this,'_module', {value: n._module, enumerable: false, writable: true })
}
if (n._path) {
Object.defineProperty(this,'_path', {value: n._path, enumerable: false, writable: true })
}
this.updateWires(n.wires);
}
util.inherits(Node, EventEmitter);
2019-08-06 15:27:56 +02:00
/**
* Update the wiring configuration for this node.
*
* We try to optimise the message handling path. To do this there are three
* cases to consider:
* 1. this node is wired to nothing. In this case we replace node.send with a
* NO-OP function.
* 2. this node is wired to one other node. In this case we set `this._wire`
* as a reference to the node it is wired to. This means we avoid unnecessary
* iterations over what would otherwise be a 1-element array.
* 3. this node is wired to multiple things. The normal node.send processing of
* this.wires applies.
*
* @param {array} wires the new wiring configuration
*/
Node.prototype.updateWires = function(wires) {
//console.log("UPDATE",this.id);
this.wires = wires || [];
delete this._wire;
var wc = 0;
this.wires.forEach(function(w) {
wc+=w.length;
});
this._wireCount = wc;
if (wc === 0) {
// With nothing wired to the node, no-op send
2019-08-06 15:27:56 +02:00
this.send = NOOP_SEND
} else {
this.send = Node.prototype.send;
if (this.wires.length === 1 && this.wires[0].length === 1) {
// Single wire, so we can shortcut the send when
// a single message is sent
this._wire = this.wires[0][0];
}
}
2014-05-03 23:26:35 +02:00
}
2019-08-06 15:27:56 +02:00
/**
* Get the context object for this node.
*
* As most nodes do not use context, this is a lazy function that will only
* create a context instance for the node if it is needed.
* @return {object} the context object
*/
2015-12-29 23:16:51 +01:00
Node.prototype.context = function() {
if (!this._context) {
this._context = context.get(this._alias||this.id,this.z);
2015-12-29 23:16:51 +01:00
}
return this._context;
}
2014-05-15 21:55:01 +02:00
2019-08-06 15:27:56 +02:00
/**
* Handle the complete event for a message
*
* @param {object} msg The message that has completed
* @param {error} error (optional) an error hit whilst handling the message
*/
2019-07-09 00:23:33 +02:00
Node.prototype._complete = function(msg,error) {
this.metric("done",msg);
hooks.trigger("onComplete",{ msg: msg, error: error, node: { id: this.id, node: this }}, err => {
if (err) {
this.error(err);
}
})
2019-07-09 00:23:33 +02:00
if (error) {
// For now, delegate this to this.error
// But at some point, the timeout handling will need to know about
// this as well.
this.error(error,msg);
} else {
this._flow.handleComplete(this,msg);
}
}
2014-05-15 21:55:01 +02:00
2019-08-06 15:27:56 +02:00
/**
* An internal reference to the original EventEmitter.on() function
*/
2014-05-15 21:55:01 +02:00
Node.prototype._on = Node.prototype.on;
2019-08-06 15:27:56 +02:00
/**
* Register a callback function for a named event.
* 'close' and 'input' events are handled locally, other events defer to EventEmitter.on()
*/
Node.prototype.on = function(event, callback) {
var node = this;
2014-05-14 22:17:54 +02:00
if (event == "close") {
this._closeCallbacks.push(callback);
2019-07-09 00:23:33 +02:00
} else if (event === "input") {
if (this._inputCallback) {
this._inputCallbacks = [this._inputCallback, callback];
this._inputCallback = null;
} else if (this._inputCallbacks) {
this._inputCallbacks.push(callback);
} else {
this._inputCallback = callback;
}
} else {
this._on(event, callback);
}
};
2019-08-06 15:27:56 +02:00
/**
* An internal reference to the original EventEmitter.emit() function
*/
2019-07-09 00:23:33 +02:00
Node.prototype._emit = Node.prototype.emit;
2019-08-06 15:27:56 +02:00
/**
* Emit an event to all registered listeners.
*/
Node.prototype.emit = function(event, ...args) {
2019-07-09 00:23:33 +02:00
var node = this;
if (event === "input") {
this._emitInput.apply(this,args);
2019-07-09 00:23:33 +02:00
} else {
this._emit.apply(this,arguments);
2019-07-09 00:23:33 +02:00
}
}
2019-08-06 15:27:56 +02:00
/**
* Handle the 'input' event.
*
* This will call all registered handlers for the 'input' event.
*/
2019-07-09 00:23:33 +02:00
Node.prototype._emitInput = function(arg) {
var node = this;
this.metric("receive", arg);
let receiveEvent = { msg:arg, destination: { id: this.id, node: this } }
// onReceive - a node is about to receive a message
hooks.trigger("onReceive",receiveEvent,(err) => {
if (err) {
node.error(err);
return
} else if (err !== false) {
if (node._inputCallback) {
// Just one callback registered.
try {
node._inputCallback(
arg,
function() { node.send.apply(node,arguments) },
function(err) { node._complete(arg,err); }
);
} catch(err) {
node.error(err,arg);
}
} else if (node._inputCallbacks) {
// Multiple callbacks registered. Call each one, tracking eventual completion
var c = node._inputCallbacks.length;
for (var i=0;i<c;i++) {
var cb = node._inputCallbacks[i];
if (cb.length === 2) {
c++;
2019-07-09 00:23:33 +02:00
}
try {
cb.call(
node,
arg,
function() { node.send.apply(node,arguments) },
function(err) {
c--;
if (c === 0) {
node._complete(arg,err);
}
}
);
} catch(err) {
node.error(err,arg);
}
}
2019-07-09 00:23:33 +02:00
}
// postReceive - the message has been passed to the node's input handler
hooks.trigger("postReceive",receiveEvent,(err) => {if (err) { node.error(err) }});
2019-07-09 00:23:33 +02:00
}
});
2019-07-09 00:23:33 +02:00
}
2019-08-06 15:27:56 +02:00
/**
* An internal reference to the original EventEmitter.removeListener() function
*/
Node.prototype._removeListener = Node.prototype.removeListener;
/**
* Remove a listener for an event
*/
Node.prototype.removeListener = function(name, listener) {
var index;
if (name === "input") {
if (this._inputCallback && this._inputCallback === listener) {
// Removing the only callback
this._inputCallback = null;
} else if (this._inputCallbacks) {
// Removing one of many callbacks
index = this._inputCallbacks.indexOf(listener);
if (index > -1) {
this._inputCallbacks.splice(index,1);
}
// Check if we can optimise back to a single callback
if (this._inputCallbacks.length === 1) {
this._inputCallback = this._inputCallbacks[0];
this._inputCallbacks = null;
}
}
} else if (name === "close") {
index = this._closeCallbacks.indexOf(listener);
if (index > -1) {
this._closeCallbacks.splice(index,1);
}
} else {
this._removeListener(name, listener);
}
}
/**
* An internal reference to the original EventEmitter.removeAllListeners() function
*/
2019-07-09 00:23:33 +02:00
Node.prototype._removeAllListeners = Node.prototype.removeAllListeners;
2019-08-06 15:27:56 +02:00
/**
* Remove all listeners for an event
*/
2019-07-09 00:23:33 +02:00
Node.prototype.removeAllListeners = function(name) {
if (name === "input") {
this._inputCallback = null;
this._inputCallbacks = null;
} else if (name === "close") {
this._closeCallbacks = [];
} else {
this._removeAllListeners(name);
}
}
2019-08-06 15:27:56 +02:00
/**
* Called when the node is being stopped
* @param {boolean} removed Whether the node has been removed, or just being stopped
* @return {Promise} resolves when the node has closed
*/
Node.prototype.close = function(removed) {
//console.log(this.type,this.id,removed);
var promises = [];
var node = this;
2019-08-06 15:27:56 +02:00
// Call all registered close callbacks.
for (var i=0;i<this._closeCallbacks.length;i++) {
var callback = this._closeCallbacks[i];
if (callback.length > 0) {
2019-08-06 15:27:56 +02:00
// The callback takes a 'done' callback and (maybe) the removed flag
promises.push(
2019-01-17 14:18:26 +01:00
new Promise((resolve) => {
try {
var args = [];
if (callback.length === 2) {
2019-08-06 15:27:56 +02:00
// The listener expects the removed flag
2019-01-17 14:18:26 +01:00
args.push(!!removed);
}
args.push(() => {
resolve();
});
callback.apply(node, args);
} catch(err) {
// TODO: error thrown in node async close callback
// We've never logged this properly.
resolve();
}
})
);
2014-05-14 22:17:54 +02:00
} else {
2019-08-06 15:27:56 +02:00
// No done callback so handle synchronously
2019-01-17 14:18:26 +01:00
try {
callback.call(node);
} catch(err) {
// TODO: error thrown in node sync close callback
// We've never logged this properly.
}
2014-05-14 22:17:54 +02:00
}
}
if (promises.length > 0) {
return Promise.all(promises).then(() => {
this.removeAllListeners("input")
2016-06-11 23:53:27 +02:00
if (this._context) {
return context.delete(this._alias||this.id,this.z);
2016-06-11 23:53:27 +02:00
}
});
2014-05-15 21:55:01 +02:00
} else {
this.removeAllListeners("input")
2016-06-11 23:53:27 +02:00
if (this._context) {
return context.delete(this._alias||this.id,this.z);
2016-06-11 23:53:27 +02:00
}
2019-01-17 14:18:26 +01:00
return Promise.resolve();
2014-05-14 22:17:54 +02:00
}
};
2019-08-06 15:27:56 +02:00
/**
* Send a message to the nodes wired.
*
*
* @param {object} msg A message or array of messages to send
*/
2014-05-03 23:26:35 +02:00
Node.prototype.send = function(msg) {
var msgSent = false;
var node;
if (msg === null || typeof msg === "undefined") {
2014-07-09 09:01:52 +02:00
return;
} else if (!util.isArray(msg)) {
// A single message has been passed in
if (typeof msg !== 'object') {
this.error(Log._("nodes.flow.non-message-returned", { type: typeof msg }));
return
}
if (this._wire) {
// A single message and a single wire on output 0
// TODO: pre-load flows.get calls - cannot do in constructor
// as not all nodes are defined at that point
if (!msg._msgid) {
msg._msgid = redUtil.generateId();
2015-01-27 15:41:20 +01:00
}
2015-02-03 23:02:26 +01:00
this.metric("send",msg);
this._flow.send([{
msg: msg,
source: {
id: this.id,
node: this,
port: 0
},
destination: {
id: this._wire,
node: undefined
},
cloneMessage: false
}]);
return;
} else {
msg = [msg];
}
2014-05-03 23:26:35 +02:00
}
var numOutputs = this.wires.length;
// Build a list of send events so that all cloning is done before
// any calls to node.receive
var sendEvents = [];
2015-02-03 23:02:26 +01:00
var sentMessageId = null;
var hasMissingIds = false;
// for each output of node eg. [msgs to output 0, msgs to output 1, ...]
for (var i = 0; i < numOutputs; i++) {
var wires = this.wires[i]; // wires leaving output i
/* istanbul ignore else */
2014-05-03 23:26:35 +02:00
if (i < msg.length) {
var msgs = msg[i]; // msgs going to output i
if (msgs !== null && typeof msgs !== "undefined") {
if (!util.isArray(msgs)) {
msgs = [msgs];
2014-05-03 23:26:35 +02:00
}
var k = 0;
// for each recipent node of that output
for (var j = 0; j < wires.length; j++) {
// for each msg to send eg. [[m1, m2, ...], ...]
for (k = 0; k < msgs.length; k++) {
var m = msgs[k];
if (m !== null && m !== undefined) {
if (typeof m !== 'object') {
this.error(Log._("nodes.flow.non-message-returned", { type: typeof m }));
} else {
if (!m._msgid) {
hasMissingIds = true;
}
/* istanbul ignore else */
if (!sentMessageId) {
sentMessageId = m._msgid;
}
sendEvents.push({
msg: m,
source: {
id: this.id,
node: this,
port: i
},
destination: {
id: wires[j],
node: undefined
},
cloneMessage: msgSent
});
msgSent = true;
}
2014-05-03 23:26:35 +02:00
}
}
}
2014-05-03 23:26:35 +02:00
}
}
}
/* istanbul ignore else */
2015-02-03 23:02:26 +01:00
if (!sentMessageId) {
sentMessageId = redUtil.generateId();
2015-02-03 23:02:26 +01:00
}
this.metric("send",{_msgid:sentMessageId});
if (hasMissingIds) {
for (i=0;i<sendEvents.length;i++) {
var ev = sendEvents[i];
/* istanbul ignore else */
if (!ev.msg._msgid) {
ev.msg._msgid = sentMessageId;
}
2015-02-03 23:02:26 +01:00
}
}
this._flow.send(sendEvents);
};
2014-05-03 23:26:35 +02:00
2019-08-06 15:27:56 +02:00
/**
* Receive a message.
*
* This will emit the `input` event with the provided message.
*/
Node.prototype.receive = function(msg) {
2015-01-27 15:41:20 +01:00
if (!msg) {
msg = {};
}
if (!msg._msgid) {
msg._msgid = redUtil.generateId();
2015-01-27 15:41:20 +01:00
}
2019-07-09 00:23:33 +02:00
this.emit("input",msg);
};
2014-05-03 23:26:35 +02:00
2014-07-09 08:42:34 +02:00
function log_helper(self, level, msg) {
var o = {
level: level,
id: self.id,
type: self.type,
msg: msg
};
if (self._alias) {
o._alias = self._alias;
}
if (self.z) {
o.z = self.z;
}
2014-07-09 08:42:34 +02:00
if (self.name) {
o.name = self.name;
2014-07-02 00:46:25 +02:00
}
// See https://github.com/node-red/node-red/issues/3327
// See https://github.com/node-red/node-red/issues/3389
let srcError;
if (msg instanceof Error) {
srcError = msg;//use existing err object for actual stack
} else {
srcError = new Error(msg);//generate a new error for generate a stack
}
try {
if(self instanceof Node && self._flow) {
self._flow.log(o);
} else {
//if self._flow is not present, this is not a node-red Node
//Set info to "Node object is not a node-red Node" to point out the `Node type` problem in log
logUnexpectedError(self, srcError, "Node object is not a node-red Node")
}
} catch(err) {
//build an unexpected error report indicating using the original error (for better stack trace)
logUnexpectedError(self, srcError, `An error occured attempting to make a log entry: ${err}`)
}
2014-05-03 23:26:35 +02:00
}
2019-08-06 15:27:56 +02:00
/**
* Log an INFO level message
*/
2014-07-09 08:42:34 +02:00
Node.prototype.log = function(msg) {
2015-02-03 23:02:26 +01:00
log_helper(this, Log.INFO, msg);
};
2014-07-09 08:42:34 +02:00
2019-08-06 15:27:56 +02:00
/**
* Log a WARN level message
*/
2014-05-03 23:26:35 +02:00
Node.prototype.warn = function(msg) {
2015-02-03 23:02:26 +01:00
log_helper(this, Log.WARN, msg);
};
2014-07-09 08:42:34 +02:00
2019-08-06 15:27:56 +02:00
/**
* Log an ERROR level message
*/
2015-02-20 02:17:24 +01:00
Node.prototype.error = function(logMessage,msg) {
if (typeof logMessage != 'boolean') {
logMessage = logMessage || "";
}
var handled = false;
if (this._flow && msg && typeof msg === 'object') {
2019-01-16 17:27:19 +01:00
handled = this._flow.handleError(this,logMessage,msg);
}
if (!handled) {
log_helper(this, Log.ERROR, logMessage);
}
};
2014-07-09 08:42:34 +02:00
2019-08-06 15:27:56 +02:00
/**
* Log an DEBUG level message
*/
Node.prototype.debug = function(msg) {
log_helper(this, Log.DEBUG, msg);
}
2019-08-06 15:27:56 +02:00
/**
* Log an TRACE level message
*/
Node.prototype.trace = function(msg) {
log_helper(this, Log.TRACE, msg);
}
2015-02-04 21:44:07 +01:00
/**
2019-08-06 15:27:56 +02:00
* Log a metric event.
2015-02-04 21:44:07 +01:00
* If called with no args, returns whether metric collection is enabled
*/
Node.prototype.metric = function(eventname, msg, metricValue) {
2015-02-04 21:44:07 +01:00
if (typeof eventname === "undefined") {
return Log.metric();
}
var metrics = {};
2015-02-03 23:02:26 +01:00
metrics.level = Log.METRIC;
2015-01-27 15:41:20 +01:00
metrics.nodeid = this.id;
metrics.event = "node."+this.type+"."+eventname;
metrics.msgid = msg._msgid;
2015-02-04 23:28:17 +01:00
metrics.value = metricValue;
2015-01-27 15:41:20 +01:00
Log.log(metrics);
}
2014-05-08 15:15:54 +02:00
/**
2019-08-06 15:27:56 +02:00
* Set the node's status object
*
2014-05-08 15:15:54 +02:00
* status: { fill:"red|green", shape:"dot|ring", text:"blah" }
2019-04-07 17:23:17 +02:00
* or
* status: "simple text status"
2014-05-08 15:15:54 +02:00
*/
Node.prototype.status = function(status) {
switch (typeof status) {
case "string":
case "number":
case "boolean":
status = {text:""+status}
}
2019-01-16 17:27:19 +01:00
this._flow.handleStatus(this,status);
};
2019-01-17 14:18:26 +01:00
function inspectObject(flow) {
try {
let properties = new Set()
let currentObj = flow
do {
if (!Object.getPrototypeOf(currentObj)) { break }
Object.getOwnPropertyNames(currentObj).map(item => properties.add(item))
} while ((currentObj = Object.getPrototypeOf(currentObj)))
let propList = [...properties.keys()].map(item => `${item}[${(typeof flow[item])[0]}]`)
propList.sort();
let result = [];
let line = "";
while (propList.length > 0) {
let prop = propList.shift()
if (line.length+prop.length > 80) {
result.push(line)
line = "";
} else {
line += " "+prop
}
}
if (line.length > 0) {
result.push(line);
}
return result.join("\n ")
} catch(err) {
return "Failed to capture object properties: "+err.toString()
}
}
function logUnexpectedError(node, error, info) {
const header = `
********************************************************************
Unexpected Node Error
********************************************************************`;
const footer = `
Please report this issue, including the information logged above:
https://github.com/node-red/node-red/issues/
********************************************************************`;
let detail = [`Info:\n ${info || 'No additional info'}`];
//Include Error info?
if(error && error.stack){
detail.push(`Stack:\n ${error.stack}`)
}
//Include Node info?
if(node && (node._module || node.type)){
const moduleInfo = node._module?`${node._module.module}@${node._module.version}`:"undefined";
const id = node._alias||node.id||"undefined";
detail.push(`Node:\n Type: ${node.type}\n Module: ${moduleInfo}\n ID: ${id}\n Properties:\n ${inspectObject(node)}`)
}
//Include Flow info?
if(node && node._flow){
detail.push(`Flow: ${node._flow.path}\n Type: ${node._flow.TYPE}\n Properties:\n ${inspectObject(node._flow)}`)
}
Log.error(`${header}\n${detail.join("\n")}\n${footer}`);
}
2014-05-03 23:26:35 +02:00
module.exports = Node;