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/Node.js

256 lines
6.9 KiB
JavaScript
Raw Normal View History

2014-05-03 23:26:35 +02:00
/**
2015-02-03 23:02:26 +01:00
* Copyright 2014, 2015 IBM Corp.
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;
2014-05-14 22:17:54 +02:00
var when = require("when");
2014-05-03 23:26:35 +02:00
var redUtil = require("../util");
2015-01-27 15:41:20 +01:00
var Log = require("../log");
2014-05-03 23:26:35 +02:00
var flows = require("./flows");
2014-05-08 15:15:54 +02:00
var comms = require("../comms");
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._closeCallbacks = [];
2014-05-03 23:26:35 +02:00
if (n.name) {
this.name = n.name;
}
this.updateWires(n.wires);
}
util.inherits(Node, EventEmitter);
Node.prototype.updateWires = function(wires) {
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
this.send = function(msg) {}
} 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
}
2014-05-15 21:55:01 +02:00
Node.prototype._on = Node.prototype.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);
} else {
this._on(event, callback);
}
};
Node.prototype.close = function() {
var promises = [];
var node = this;
for (var i=0;i<this._closeCallbacks.length;i++) {
var callback = this._closeCallbacks[i];
2014-05-14 22:17:54 +02:00
if (callback.length == 1) {
promises.push(
when.promise(function(resolve) {
callback.call(node, function() {
2014-05-14 22:17:54 +02:00
resolve();
});
})
);
2014-05-14 22:17:54 +02:00
} else {
callback.call(node);
2014-05-14 22:17:54 +02:00
}
}
if (promises.length > 0) {
return when.settle(promises);
2014-05-15 21:55:01 +02:00
} else {
return;
2014-05-14 22:17:54 +02:00
}
};
2015-01-27 15:41:20 +01:00
function constructUniqueIdentifier() {
return (1+Math.random()*4294967295).toString(16);
}
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)) {
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 = constructUniqueIdentifier();
2015-01-27 15:41:20 +01:00
}
2015-02-03 23:02:26 +01:00
this.metric("send",msg);
node = flows.get(this._wire);
/* istanbul ignore else */
if (node) {
node.receive(msg);
}
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;
// 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++) {
node = flows.get(wires[j]); // node at end of wire j
if (node) {
// for each msg to send eg. [[m1, m2, ...], ...]
for (k = 0; k < msgs.length; k++) {
2015-02-03 23:02:26 +01:00
var m = msgs[k];
/* istanbul ignore else */
2015-02-03 23:02:26 +01:00
if (!sentMessageId) {
sentMessageId = m._msgid;
2015-02-03 23:02:26 +01:00
}
if (msgSent) {
2015-02-03 23:02:26 +01:00
var clonedmsg = redUtil.cloneMessage(m);
2015-01-27 15:41:20 +01:00
sendEvents.push({n:node,m:clonedmsg});
} else {
2015-02-03 23:02:26 +01:00
sendEvents.push({n:node,m:m});
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 = constructUniqueIdentifier();
}
this.metric("send",{_msgid:sentMessageId});
for (i=0;i<sendEvents.length;i++) {
var ev = sendEvents[i];
/* istanbul ignore else */
if (!ev.m._msgid) {
ev.m._msgid = sentMessageId;
2015-02-03 23:02:26 +01:00
}
ev.n.receive(ev.m);
}
};
2014-05-03 23:26:35 +02:00
Node.prototype.receive = function(msg) {
2015-01-27 15:41:20 +01:00
if (!msg) {
msg = {};
}
if (!msg._msgid) {
msg._msgid = constructUniqueIdentifier();
2015-01-27 15:41:20 +01:00
}
this.metric("receive",msg);
try {
this.emit("input", msg);
} catch(err) {
this.error(err,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
};
2014-07-09 08:42:34 +02:00
if (self.name) {
o.name = self.name;
2014-07-02 00:46:25 +02:00
}
2015-01-27 15:41:20 +01:00
Log.log(o);
2014-05-03 23:26:35 +02:00
}
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
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
2015-02-20 02:17:24 +01:00
Node.prototype.error = function(logMessage,msg) {
logMessage = logMessage || "";
log_helper(this, Log.ERROR, logMessage);
/* istanbul ignore else */
if (msg) {
flows.handleError(this,logMessage,msg);
}
};
2014-07-09 08:42:34 +02:00
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
/**
* status: { fill:"red|green", shape:"dot|ring", text:"blah" }
*/
Node.prototype.status = function(status) {
comms.publish("status/" + this.id, status, true);
};
2014-05-03 23:26:35 +02:00
module.exports = Node;