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/Flow.js
Nick O'Leary f626ee060a Allow properties to be specified by environment variables
A property set to $(ABC) will be substituted with the environment
variable ABC - if it exists. If the property doesn't exist, the property
is left unchanged.
2015-10-23 22:14:39 +01:00

846 lines
27 KiB
JavaScript

/**
* Copyright 2015 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 when = require("when");
var clone = require("clone");
var typeRegistry = require("./registry");
var credentials = require("./credentials");
var redUtil = require("../util");
var events = require("../events");
var Log = require("../log");
function getID() {
return (1+Math.random()*4294967295).toString(16);
}
var EnvVarPropertyRE = /^\$\((\S+)\)$/;
function mapEnvVarProperties(obj,prop) {
if (Buffer.isBuffer(obj[prop])) {
return;
} else if (Array.isArray(obj[prop])) {
for (var i=0;i<obj[prop].length;i++) {
mapEnvVarProperties(obj[prop],i);
}
} else if (typeof obj[prop] === 'string') {
var m;
if ( (m = EnvVarPropertyRE.exec(obj[prop])) !== null) {
if (process.env.hasOwnProperty(m[1])) {
obj[prop] = process.env[m[1]];
}
}
} else {
for (var p in obj[prop]) {
if (obj[prop].hasOwnProperty) {
mapEnvVarProperties(obj[prop],p);
}
}
}
}
function createNode(type,config) {
var nn = null;
var m;
var nt = typeRegistry.get(type);
if (nt) {
for (var p in config) {
if (config.hasOwnProperty(p)) {
mapEnvVarProperties(config,p);
}
}
try {
nn = new nt(clone(config));
}
catch (err) {
Log.log({
level: Log.ERROR,
id:config.id,
type: type,
msg: err
});
}
} else {
Log.error(Log._("nodes.flow.unknown-type", {type:type}));
}
return nn;
}
function createSubflow(sf,sfn,subflows) {
//console.log("CREATE SUBFLOW",sf.config.id,sfn.id);
var nodes = [];
var node_map = {};
var newNodes = [];
var node;
var wires;
var i,j,k;
// Clone all of the subflow node definitions and give them new IDs
for (i=0;i<sf.nodes.length;i++) {
node = clone(sf.nodes[i].config);
var nid = getID();
node_map[node.id] = node;
node._alias = node.id;
node.id = nid;
node.z = sfn.id;
newNodes.push(node);
}
// Look for any catch/status nodes and update their scope ids
// Update all subflow interior wiring to reflect new node IDs
for (i=0;i<newNodes.length;i++) {
node = newNodes[i];
var outputs = node.wires;
for (j=0;j<outputs.length;j++) {
wires = outputs[j];
for (k=0;k<wires.length;k++) {
outputs[j][k] = node_map[outputs[j][k]].id
}
}
if ((node.type === 'catch' || node.type === 'status') && node.scope) {
node.scope = node.scope.map(function(id) {
return node_map[id]?node_map[id].id:""
})
}
}
// Create a subflow node to accept inbound messages and route appropriately
var Node = require("./Node");
var subflowInstance = {
id: sfn.id,
type: sfn.type,
z: sfn.z,
name: sfn.name,
wires: []
}
if (sf.config.in) {
subflowInstance.wires = sf.config.in.map(function(n) { return n.wires.map(function(w) { return node_map[w.id].id;})})
subflowInstance._originalWires = clone(subflowInstance.wires);
}
var subflowNode = new Node(subflowInstance);
subflowNode.on("input", function(msg) { this.send(msg);});
subflowNode._updateWires = subflowNode.updateWires;
subflowNode.updateWires = function(newWires) {
// Wire the subflow outputs
if (sf.config.out) {
var node,wires,i,j;
// Restore the original wiring to the internal nodes
subflowInstance.wires = clone(subflowInstance._originalWires);
for (i=0;i<sf.config.out.length;i++) {
wires = sf.config.out[i].wires;
for (j=0;j<wires.length;j++) {
if (wires[j].id != sf.config.id) {
node = node_map[wires[j].id];
if (node._originalWires) {
node.wires = clone(node._originalWires);
}
}
}
}
var modifiedNodes = {};
var subflowInstanceModified = false;
for (i=0;i<sf.config.out.length;i++) {
wires = sf.config.out[i].wires;
for (j=0;j<wires.length;j++) {
if (wires[j].id === sf.config.id) {
subflowInstance.wires[wires[j].port] = subflowInstance.wires[wires[j].port].concat(newWires[i]);
subflowInstanceModified = true;
} else {
node = node_map[wires[j].id];
node.wires[wires[j].port] = node.wires[wires[j].port].concat(newWires[i]);
modifiedNodes[node.id] = node;
}
}
}
Object.keys(modifiedNodes).forEach(function(id) {
var node = modifiedNodes[id];
subflowNode.instanceNodes[id].updateWires(node.wires);
});
if (subflowInstanceModified) {
subflowNode._updateWires(subflowInstance.wires);
}
}
}
nodes.push(subflowNode);
// Wire the subflow outputs
if (sf.config.out) {
var modifiedNodes = {};
for (i=0;i<sf.config.out.length;i++) {
wires = sf.config.out[i].wires;
for (j=0;j<wires.length;j++) {
if (wires[j].id === sf.config.id) {
// A subflow input wired straight to a subflow output
subflowInstance.wires[wires[j].port] = subflowInstance.wires[wires[j].port].concat(sfn.wires[i])
subflowNode._updateWires(subflowInstance.wires);
} else {
node = node_map[wires[j].id];
modifiedNodes[node.id] = node;
if (!node._originalWires) {
node._originalWires = clone(node.wires);
}
node.wires[wires[j].port] = (node.wires[wires[j].port]||[]).concat(sfn.wires[i]);
}
}
}
}
// Instantiate the nodes
for (i=0;i<newNodes.length;i++) {
node = newNodes[i];
var type = node.type;
var m = /^subflow:(.+)$/.exec(type);
if (!m) {
nodes.push(createNode(type,node));
} else {
var subflowId = m[1];
nodes = nodes.concat(createSubflow(subflows[subflowId],node,subflows));
}
}
subflowNode.instanceNodes = {};
nodes.forEach(function(node) {
subflowNode.instanceNodes[node.id] = node;
});
return nodes;
}
function diffNodeConfigs(oldNode,newNode) {
if (oldNode == null) {
return true;
} else {
for (var p in newNode) {
if (newNode.hasOwnProperty(p) && p != "x" && p != "y" && p != "wires") {
if (!redUtil.compareObjects(oldNode[p],newNode[p])) {
return true;
}
}
}
}
return false;
}
var subflowInstanceRE = /^subflow:(.+)$/;
function Flow(config) {
this.activeNodes = {};
this.subflowInstanceNodes = {};
this.catchNodeMap = {};
this.statusNodeMap = {};
this.started = false;
this.parseConfig(config);
}
Flow.prototype.parseConfig = function(config) {
var i;
var nodeConfig;
var nodeType;
this.config = config;
this.allNodes = {};
this.nodes = {};
this.subflows = {};
this.configNodes = {};
var unknownTypes = {};
for (i=0;i<this.config.length;i++) {
nodeConfig = this.config[i];
nodeType = nodeConfig.type;
this.allNodes[nodeConfig.id] = nodeConfig;
if (nodeType == "subflow") {
this.subflows[nodeConfig.id] = {
type: "subflow",
config: nodeConfig,
nodes: []
}
}
}
//console.log("Known subflows:",Object.keys(this.subflows));
for (i=0;i<this.config.length;i++) {
nodeConfig = this.config[i];
nodeType = nodeConfig.type;
if (nodeConfig.credentials) {
delete nodeConfig.credentials;
}
if (nodeType != "tab" && nodeType != "subflow") {
var m = subflowInstanceRE.exec(nodeType);
if ((m && !this.subflows[m[1]]) || (!m && !typeRegistry.get(nodeType))) {
// This is an unknown subflow or an unknown type
unknownTypes[nodeType] = true;
} else {
var nodeInfo = {
type: nodeType,
config:nodeConfig
}
if (m) {
nodeInfo.subflow = m[1];
}
if (this.subflows[nodeConfig.z]) {
this.subflows[nodeConfig.z].nodes.push(nodeInfo);
} else {
this.nodes[nodeConfig.id] = nodeInfo;
}
if (nodeConfig.type != "catch" && nodeConfig.type != "status") {
for (var prop in nodeConfig) {
if (nodeConfig.hasOwnProperty(prop) &&
prop != "type" &&
prop != "id" &&
prop != "z" &&
prop != "wires" &&
this.allNodes[nodeConfig[prop]]) {
this.configNodes[nodeConfig[prop]] = this.allNodes[nodeConfig[prop]];
}
}
}
}
}
}
//console.log("NODES");
//for (i in this.nodes) {
// if (this.nodes.hasOwnProperty(i)) {
// console.log(" ",i,this.nodes[i].type,this.nodes[i].config.name||"");
// }
//}
//console.log("SUBFLOWS");
//for (i in this.subflows) {
// if (this.subflows.hasOwnProperty(i)) {
// console.log(" ",i,this.subflows[i].type,this.subflows[i].config.name||"");
// for (var j=0;j<this.subflows[i].nodes.length;j++) {
// console.log(" ",this.subflows[i].nodes[j].config.id,this.subflows[i].nodes[j].type,this.subflows[i].nodes[j].config.name||"");
// }
// }
//}
this.missingTypes = Object.keys(unknownTypes);
}
Flow.prototype.start = function(configDiff) {
if (configDiff) {
for (var j=0;j<configDiff.rewire.length;j++) {
var rewireNode = this.activeNodes[configDiff.rewire[j]];
if (rewireNode) {
rewireNode.updateWires(this.allNodes[rewireNode.id].wires);
}
}
}
this.started = true;
if (this.missingTypes.length > 0) {
throw new Error(Log._("nodes.flow.missing-types"));
}
events.emit("nodes-starting");
var id;
var node;
for (id in this.configNodes) {
if (this.configNodes.hasOwnProperty(id)) {
node = this.configNodes[id];
if (!this.activeNodes[id]) {
this.activeNodes[id] = createNode(node.type,node);
}
}
}
for (id in this.nodes) {
if (this.nodes.hasOwnProperty(id)) {
node = this.nodes[id];
if (!node.subflow) {
if (!this.activeNodes[id]) {
this.activeNodes[id] = createNode(node.type,node.config);
//console.log(id,"created");
} else {
//console.log(id,"already running");
}
} else {
if (!this.subflowInstanceNodes[id]) {
var nodes = createSubflow(this.subflows[node.subflow],node.config,this.subflows);
this.subflowInstanceNodes[id] = nodes.map(function(n) { return n.id});
for (var i=0;i<nodes.length;i++) {
this.activeNodes[nodes[i].id] = nodes[i];
}
//console.log(id,"(sf) created");
} else {
//console.log(id,"(sf) already running");
}
}
}
}
this.catchNodeMap = {};
this.statusNodeMap = {};
for (id in this.activeNodes) {
if (this.activeNodes.hasOwnProperty(id)) {
node = this.activeNodes[id];
if (node.type === "catch") {
this.catchNodeMap[node.z] = this.catchNodeMap[node.z] || [];
this.catchNodeMap[node.z].push(node);
} else if (node.type === "status") {
this.statusNodeMap[node.z] = this.statusNodeMap[node.z] || [];
this.statusNodeMap[node.z].push(node);
}
}
}
credentials.clean(this.config);
events.emit("nodes-started");
}
Flow.prototype.stop = function(configDiff) {
var nodeList;
if (configDiff) {
nodeList = configDiff.stop;
} else {
nodeList = Object.keys(this.activeNodes);
}
var flow = this;
return when.promise(function(resolve) {
events.emit("nodes-stopping");
var promises = [];
for (var i=0;i<nodeList.length;i++) {
var node = flow.activeNodes[nodeList[i]];
if (node) {
try {
var p = node.close();
if (p) {
promises.push(p);
}
} catch(err) {
node.error(err);
}
delete flow.subflowInstanceNodes[nodeList[i]];
delete flow.activeNodes[nodeList[i]];
}
}
when.settle(promises).then(function() {
events.emit("nodes-stopped");
flow.started = false;
resolve();
});
});
}
Flow.prototype.getMissingTypes = function() {
return this.missingTypes;
}
Flow.prototype.typeRegistered = function(type) {
if (this.missingTypes.length > 0) {
var i = this.missingTypes.indexOf(type);
if (i != -1) {
this.missingTypes.splice(i,1);
if (this.missingTypes.length === 0 && this.started) {
this.start();
}
return true;
}
}
return false;
}
Flow.prototype.getNode = function(id) {
return this.activeNodes[id];
}
Flow.prototype.getFlow = function() {
//console.log(this.config);
return this.config;
}
Flow.prototype.eachNode = function(callback) {
for (var id in this.activeNodes) {
if (this.activeNodes.hasOwnProperty(id)) {
callback(this.activeNodes[id]);
}
}
}
Flow.prototype.diffConfig = function(config,type) {
var activeNodesToStop = [];
var nodesToRewire = [];
if (type && type!="full") {
var diff = diffFlow(this,config);
//var diff = {
// deleted:[]
// changed:[]
// linked:[]
// wiringChanged: []
//}
var nodesToStop = [];
nodesToRewire = diff.wiringChanged;
if (type == "nodes") {
nodesToStop = diff.deleted.concat(diff.changed);
} else if (type == "flows") {
nodesToStop = diff.deleted.concat(diff.changed).concat(diff.linked);
}
for (var i=0;i<nodesToStop.length;i++) {
var id = nodesToStop[i];
if (this.subflowInstanceNodes[id]) {
activeNodesToStop = activeNodesToStop.concat(this.subflowInstanceNodes[id]);
} else if (this.activeNodes[id]) {
activeNodesToStop.push(id);
}
}
} else {
activeNodesToStop = Object.keys(this.activeNodes);
}
return {
type: type,
stop: activeNodesToStop,
rewire: nodesToRewire,
config: config
}
}
function diffFlow(flow,config) {
//if (!flow.started) {
// throw new Error("Cannot diff an unstarted flow");
//}
var flowNodes = {};
var changedNodes = {};
var deletedNodes = {};
var deletedSubflows = {};
var deletedTabs = {};
var linkChangedNodes = {};
var activeLinks = {};
var newLinks = {};
var changedSubflowStack = [];
var changedSubflows = {};
var buildNodeLinks = function(nodeLinks,n,nodes) {
nodeLinks[n.id] = nodeLinks[n.id] || [];
if (n.wires) {
for (var j=0;j<n.wires.length;j++) {
var wires = n.wires[j];
for (var k=0;k<wires.length;k++) {
nodeLinks[n.id].push(wires[k]);
var nn = nodes[wires[k]];
if (nn) {
nodeLinks[nn.id] = nodeLinks[nn.id] || [];
nodeLinks[nn.id].push(n.id);
}
}
}
}
}
config.forEach(function(node) {
flowNodes[node.id] = node;
});
config.forEach(function(node) {
var changed = false;
if (node.credentials) {
changed = true;
delete node.credentials;
} else {
changed = diffNodeConfigs(flow.allNodes[node.id],node);
if (!changed) {
if (flowNodes[node.z] && flowNodes[node.z].type == "subflow") {
var originalNode = flow.allNodes[node.id];
if (originalNode && !redUtil.compareObjects(originalNode.wires,node.wires)) {
// This is a node in a subflow whose wiring has changed. Mark subflow as changed
changed = true;
}
}
}
}
if (changed) {
changedNodes[node.id] = node;
}
});
flow.config.forEach(function(node) {
if (!flowNodes[node.id]) {
if (node.type === "tab") {
deletedTabs[node.id] = node;
} else if (node.type === "subflow") {
deletedSubflows[node.id] = node;
} else {
deletedNodes[node.id] = node;
}
}
buildNodeLinks(activeLinks,node,flow.allNodes);
});
flow.config.forEach(function(node) {
for (var prop in node) {
if (node.hasOwnProperty(prop) && prop != "z" && prop != "id" && prop != "wires") {
// This node has a property that references a changed node
// Assume it is a config node change and mark this node as
// changed.
if (changedNodes[node[prop]]) {
changedNodes[node.id] = node;
}
}
}
});
var checkSubflowMembership = function(nodes,id) {
var node = nodes[id];
if (node) {
if (node.type == "subflow") {
changedSubflows[id] = node;
changedSubflowStack.push(id);
} else if (nodes[node.z] && nodes[node.z].type == "subflow") {
if (!changedSubflows[node.z]) {
changedSubflows[node.z] = nodes[node.z];
changedSubflowStack.push(node.z);
}
}
}
};
Object.keys(changedNodes).forEach(function(n) { checkSubflowMembership(flowNodes,n)});
Object.keys(deletedNodes).forEach(function(n) { checkSubflowMembership(flow.allNodes,n)});
while (changedSubflowStack.length > 0) {
var subflowId = changedSubflowStack.pop();
config.forEach(function(node) {
if (node.type == "subflow:"+subflowId) {
if (!changedNodes[node.id]) {
changedNodes[node.id] = node;
checkSubflowMembership(flowNodes,node.id);
}
}
});
}
config.forEach(function(node) {
buildNodeLinks(newLinks,node,flowNodes);
});
var markLinkedNodes = function(linkChanged,otherChangedNodes,linkMap,allNodes) {
var stack = Object.keys(changedNodes).concat(Object.keys(otherChangedNodes));
var visited = {};
while(stack.length > 0) {
var id = stack.pop();
var linkedNodes = linkMap[id];
if (linkedNodes) {
for (var i=0;i<linkedNodes.length;i++) {
var linkedNodeId = linkedNodes[i];
if (changedNodes[linkedNodeId] || deletedNodes[linkedNodeId] || otherChangedNodes[linkedNodeId] || linkChanged[linkedNodeId]) {
// Do nothing - this linked node is already marked as changed, so will get done
} else {
linkChanged[linkedNodeId] = true;
stack.push(linkedNodeId);
}
}
}
}
}
markLinkedNodes(linkChangedNodes,{},newLinks,flowNodes);
markLinkedNodes(linkChangedNodes,{},activeLinks,flow.allNodes);
var modifiedLinkNodes = {};
config.forEach(function(node) {
if (!changedNodes[node.id]) {
// only concerned about unchanged nodes whose wiring may have changed
var newNodeLinks = newLinks[node.id];
var oldNodeLinks = activeLinks[node.id];
var newLinkMap = {};
newNodeLinks.forEach(function(l) { newLinkMap[l] = (newLinkMap[l]||0)+1;});
var oldLinkMap = {};
oldNodeLinks.forEach(function(l) { oldLinkMap[l] = (oldLinkMap[l]||0)+1;});
newNodeLinks.forEach(function(link) {
if (newLinkMap[link] != oldLinkMap[link]) {
modifiedLinkNodes[node.id] = node;
linkChangedNodes[node.id] = node;
if (!changedNodes[link] && !deletedNodes[link]) {
modifiedLinkNodes[link] = flowNodes[link];
linkChangedNodes[link] = flowNodes[link];
}
}
});
oldNodeLinks.forEach(function(link) {
if (newLinkMap[link] != oldLinkMap[link]) {
modifiedLinkNodes[node.id] = node;
linkChangedNodes[node.id] = node;
if (!changedNodes[link] && !deletedNodes[link]) {
modifiedLinkNodes[link] = flowNodes[link];
linkChangedNodes[link] = flowNodes[link];
}
}
});
}
});
markLinkedNodes(linkChangedNodes,modifiedLinkNodes,newLinks,flowNodes);
// config.forEach(function(n) {
// console.log((changedNodes[n.id]!=null)?"[C]":"[ ]",(linkChangedNodes[n.id]!=null)?"[L]":"[ ]","[ ]",n.id,n.type,n.name);
// });
//
// Object.keys(deletedNodes).forEach(function(id) {
// var n = flow.allNodes[id];
// console.log("[ ] [ ] [D]",n.id,n.type);
// });
var diff = {
deleted: Object.keys(deletedNodes).filter(function(id) { return deletedNodes[id].type != "subflow" && (!deletedNodes[id].z || deletedTabs[deletedNodes[id].z] || !(deletedSubflows[deletedNodes[id].z] || flowNodes[deletedNodes[id].z].type == "subflow"))}),
changed: Object.keys(changedNodes).filter(function(id) { return changedNodes[id].type != "subflow" && (!changedNodes[id].z || flowNodes[changedNodes[id].z].type != "subflow")}),
linked: Object.keys(linkChangedNodes).filter(function(id) { return linkChangedNodes[id].type != "subflow" && (!linkChangedNodes[id].z || flowNodes[linkChangedNodes[id].z].type != "subflow")}),
wiringChanged: []
}
config.forEach(function(n) {
if (!flowNodes[n.z] || flowNodes[n.z].type != "subflow") {
var originalNode = flow.allNodes[n.id];
if (originalNode && !redUtil.compareObjects(originalNode.wires,n.wires)) {
diff.wiringChanged.push(n.id);
}
}
});
return diff;
}
Flow.prototype.handleStatus = function(node,statusMessage) {
var targetStatusNodes = null;
var reportingNode = node;
var handled = false;
while(reportingNode && !handled) {
targetStatusNodes = this.statusNodeMap[reportingNode.z];
if (targetStatusNodes) {
targetStatusNodes.forEach(function(targetStatusNode) {
if (targetStatusNode.scope && targetStatusNode.scope.indexOf(node.id) === -1) {
return;
}
var message = {
status: {
text: "",
source: {
id: node.id,
type: node.type,
name: node.name
}
}
};
if (statusMessage.text) {
message.status.text = statusMessage.text;
}
targetStatusNode.receive(message);
handled = true;
});
}
if (!handled) {
reportingNode = this.activeNodes[reportingNode.z];
}
}
}
Flow.prototype.handleError = function(node,logMessage,msg) {
var count = 1;
if (msg && msg.hasOwnProperty("error")) {
if (msg.error.hasOwnProperty("source")) {
if (msg.error.source.id === node.id) {
count = msg.error.source.count+1;
if (count === 10) {
node.warn(Log._("nodes.flow.error-loop"));
return;
}
}
}
}
var targetCatchNodes = null;
var throwingNode = node;
var handled = false;
while (throwingNode && !handled) {
targetCatchNodes = this.catchNodeMap[throwingNode.z];
if (targetCatchNodes) {
targetCatchNodes.forEach(function(targetCatchNode) {
if (targetCatchNode.scope && targetCatchNode.scope.indexOf(node.id) === -1) {
return;
}
var errorMessage;
if (msg) {
errorMessage = redUtil.cloneMessage(msg);
} else {
errorMessage = {};
}
if (errorMessage.hasOwnProperty("error")) {
errorMessage._error = errorMessage.error;
}
errorMessage.error = {
message: logMessage.toString(),
source: {
id: node.id,
type: node.type,
count: count
}
};
targetCatchNode.receive(errorMessage);
handled = true;
});
}
if (!handled) {
throwingNode = this.activeNodes[throwingNode.z];
}
}
}
module.exports = Flow;