frodovdr/node-red
1
0
Fork 0
mirror of https://github.com/node-red/node-red.git synced 2025-03-01 10:36:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor Subflow logic into own class

Browse Source
This commit is contained in:
Nick O'Leary
2019-01-16 16:27:19 +00:00
parent da756fa568
commit 81f4e0de56
9 changed files with 1214 additions and 1155 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
packages/node_modules/@node-red/runtime/lib/nodes/Node.js vendored
View File

@@ -35,6 +35,9 @@ function Node(n) {
if (n._alias) {
this._alias = n._alias;
}
if (n._flow) {
this._flow = n._flow;
}
this.updateWires(n.wires);
}
@@ -131,10 +134,12 @@ Node.prototype.send = function(msg) {
msg._msgid = redUtil.generateId();
}
this.metric("send",msg);
node = flows.get(this._wire);
node = this._flow.getNode(this._wire);
/* istanbul ignore else */
if (node) {
node.receive(msg);
} else {
console.log("trying to send to a node not on this flow",this._wire);
}
return;
} else {
@@ -163,7 +168,7 @@ Node.prototype.send = function(msg) {
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
node = this._flow.getNode(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++) {
@@ -182,6 +187,8 @@ Node.prototype.send = function(msg) {
}
}
}
} else {
console.log("trying to send to a node not on this flow",this._wire);
}
}
}
@@ -251,7 +258,7 @@ Node.prototype.error = function(logMessage,msg) {
}
var handled = false;
if (msg) {
handled = flows.handleError(this,logMessage,msg);
handled = this._flow.handleError(this,logMessage,msg);
}
if (!handled) {
log_helper(this, Log.ERROR, logMessage);
@@ -286,6 +293,6 @@ Node.prototype.metric = function(eventname, msg, metricValue) {
* status: { fill:"red|green", shape:"dot|ring", text:"blah" }
*/
Node.prototype.status = function(status) {
flows.handleStatus(this,status);
this._flow.handleStatus(this,status);
};
module.exports = Node;

544
packages/node_modules/@node-red/runtime/lib/nodes/flows/Flow.js vendored
View File

@@ -16,33 +16,68 @@
var when = require("when");
var clone = require("clone");
var typeRegistry = require("@node-red/registry");
var Subflow;
var Log;
var redUtil = require("@node-red/util").util;
var flowUtil = require("./util");
var events = require("../../events");
var nodeCloseTimeout = 15000;
class Flow {
constructor(global,flow) {
this.global = global;
constructor(parent,globalFlow,flow) {
this.TYPE = 'flow';
this.parent = parent;
this.global = globalFlow;
if (typeof flow === 'undefined') {
this.flow = global;
this.flow = globalFlow;
this.isGlobalFlow = true;
} else {
this.flow = flow;
this.isGlobalFlow = false;
}
this.id = this.flow.id || "global";
this.activeNodes = {};
this.subflowInstanceNodes = {};
this.catchNodeMap = {};
this.statusNodeMap = {};
this.catchNodes = [];
this.statusNodes = [];
}
debug(msg) {
Log.log({
id: this.id||"global",
level: Log.DEBUG,
type:this.TYPE,
msg:msg
})
}
log(msg) {
Log.log({
id: this.id||"global",
level: Log.INFO,
type:this.TYPE,
msg:msg
})
}
trace(msg) {
Log.log({
id: this.id||"global",
level: Log.TRACE,
type:this.TYPE,
msg:msg
})
}
start(diff) {
this.trace("start "+this.TYPE);
var node;
var newNode;
var id;
this.catchNodeMap = {};
this.statusNodeMap = {};
this.catchNodes = [];
this.statusNodes = [];
var configNodes = Object.keys(this.flow.configs);
var configNodeAttempts = {};
@@ -69,7 +104,7 @@ class Flow {
}
}
if (readyToCreate) {
newNode = createNode(node.type,node);
newNode = flowUtil.createNode(this,node);
if (newNode) {
this.activeNodes[id] = newNode;
}
@@ -91,7 +126,7 @@ class Flow {
node = this.flow.nodes[id];
if (!node.subflow) {
if (!this.activeNodes[id]) {
newNode = createNode(node.type,node);
newNode = flowUtil.createNode(this,node);
if (newNode) {
this.activeNodes[id] = newNode;
}
@@ -100,13 +135,24 @@ class Flow {
if (!this.subflowInstanceNodes[id]) {
try {
var subflowDefinition = this.flow.subflows[node.subflow]||this.global.subflows[node.subflow]
var nodes = createSubflow(subflowDefinition,node,this.flow.subflows,this.global.subflows,this.activeNodes);
this.subflowInstanceNodes[id] = nodes.map(function(n) { return n.id});
for (var i=0;i<nodes.length;i++) {
if (nodes[i]) {
this.activeNodes[nodes[i].id] = nodes[i];
}
}
// console.log("NEED TO CREATE A SUBFLOW",id,node.subflow);
this.subflowInstanceNodes[id] = true;
var subflow = Subflow.create(
this,
this.global,
subflowDefinition,
node
);
subflow.start();
this.activeNodes[id] = subflow.node;
this.subflowInstanceNodes[id] = subflow;
// this.subflowInstanceNodes[id] = nodes.map(function(n) { return n.id});
// for (var i=0;i<nodes.length;i++) {
// if (nodes[i]) {
// this.activeNodes[nodes[i].id] = nodes[i];
// }
// }
} catch(err) {
console.log(err.stack)
}
@@ -115,31 +161,43 @@ class Flow {
}
}
var activeCount = Object.keys(this.activeNodes).length;
if (activeCount > 0) {
this.trace("------------------|--------------|-----------------");
this.trace(" id | type | alias");
this.trace("------------------|--------------|-----------------");
}
// Build the map of catch/status nodes.
for (id in this.activeNodes) {
if (this.activeNodes.hasOwnProperty(id)) {
node = this.activeNodes[id];
this.trace(" "+id.padEnd(16)+" | "+node.type.padEnd(12)+" | "+(node._alias||""));
if (node.type === "catch") {
this.catchNodeMap[node.z] = this.catchNodeMap[node.z] || [];
this.catchNodeMap[node.z].push(node);
this.catchNodes.push(node);
} else if (node.type === "status") {
this.statusNodeMap[node.z] = this.statusNodeMap[node.z] || [];
this.statusNodeMap[node.z].push(node);
this.statusNodes.push(node);
}
}
}
if (activeCount > 0) {
this.trace("------------------|--------------|-----------------");
}
// this.dump();
}
stop(stopList, removedList) {
return new Promise((resolve,reject) => {
this.trace("stop "+this.TYPE);
var i;
if (stopList) {
for (i=0;i<stopList.length;i++) {
if (this.subflowInstanceNodes[stopList[i]]) {
// The first in the list is the instance node we already
// know about
stopList = stopList.concat(this.subflowInstanceNodes[stopList[i]].slice(1))
}
}
// for (i=0;i<stopList.length;i++) {
// if (this.subflowInstanceNodes[stopList[i]]) {
// console.log("NEED TO STOP A SUBFLOW",stopList[i]);
// // The first in the list is the instance node we already
// // know about
// // stopList = stopList.concat(this.subflowInstanceNodes[stopList[i]].slice(1))
// }
// }
} else {
stopList = Object.keys(this.activeNodes);
}
@@ -156,34 +214,20 @@ class Flow {
if (node) {
delete this.activeNodes[stopList[i]];
if (this.subflowInstanceNodes[stopList[i]]) {
try {
var subflow = this.subflowInstanceNodes[stopList[i]];
promises.push(this.stopNode(node,false).then(() => { subflow.stop() }));
} catch(err) {
node.error(err);
}
delete this.subflowInstanceNodes[stopList[i]];
}
try {
var removed = removedMap[stopList[i]];
promises.push(
when.promise(function(resolve, reject) {
var start;
var nt = node.type;
var nid = node.id;
var n = node;
when.promise(function(resolve) {
Log.trace("Stopping node "+nt+":"+nid+(removed?" removed":""));
start = Date.now();
resolve(n.close(removed));
}).timeout(nodeCloseTimeout).then(function(){
var delta = Date.now() - start;
Log.trace("Stopped node "+nt+":"+nid+" ("+delta+"ms)" );
resolve(delta);
},function(err) {
var delta = Date.now() - start;
n.error(Log._("nodes.flows.stopping-error",{message:err}));
Log.debug(err.stack);
reject(err);
});
})
);
} catch(err) {
node.error(err);
} else {
try {
var removed = removedMap[stopList[i]];
promises.push(this.stopNode(node,removed));
} catch(err) {
node.error(err);
}
}
}
}
@@ -193,13 +237,46 @@ class Flow {
});
}
stopNode(node,removed) {
return when.promise(function(resolve, reject) {
var start;
when.promise(function(resolve) {
Log.trace("Stopping node "+node.type+":"+node.id+(removed?" removed":""));
start = Date.now();
resolve(node.close(removed));
}).timeout(nodeCloseTimeout).then(function(){
var delta = Date.now() - start;
Log.trace("Stopped node "+node.type+":"+node.id+" ("+delta+"ms)" );
resolve(delta);
},function(err) {
var delta = Date.now() - start;
node.error(Log._("nodes.flows.stopping-error",{message:err}));
Log.debug(err.stack);
reject(err);
});
})
}
update(_global,_flow) {
this.global = _global;
this.flow = _flow;
}
getNode(id) {
return this.activeNodes[id];
// console.log('getNode',id,!!this.activeNodes[id])
if (!id) {
return undefined;
}
// console.log((new Error().stack).toString().split("\n").slice(1,3).join("\n"))
if ((this.flow.configs && this.flow.configs[id]) || (this.flow.nodes && this.flow.nodes[id])) {
// This is a node owned by this flow, so return whatever we have got
// During a stop/restart, activeNodes could be null for this id
return this.activeNodes[id];
} else if (this.activeNodes[id]) {
// TEMP: this is a subflow internal node within this flow
return this.activeNodes[id];
}
return this.parent.getNode(id);
}
getActiveNodes() {
@@ -207,40 +284,40 @@ class Flow {
}
handleStatus(node,statusMessage) {
var targetStatusNodes = null;
var reportingNode = node;
events.emit("node-status",{
id: node.id,
status:statusMessage
});
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.hasOwnProperty("text")) {
message.status.text = statusMessage.text.toString();
}
targetStatusNode.receive(message);
handled = true;
});
this.statusNodes.forEach(function(targetStatusNode) {
if (targetStatusNode.scope && targetStatusNode.scope.indexOf(node.id) === -1) {
return;
}
if (!handled) {
reportingNode = this.activeNodes[reportingNode.z];
var message = {
status: {
text: "",
source: {
id: node.id,
type: node.type,
name: node.name
}
}
};
if (statusMessage.hasOwnProperty("text")) {
message.status.text = statusMessage.text.toString();
}
targetStatusNode.receive(message);
handled = true;
});
if (!handled) {
// // Nothing in this flow handled the status - pass it to the parent
this.parent.handleStatus(node,statusMessage);
}
}
handleError(node,logMessage,msg) {
// console.log("HE",logMessage);
var count = 1;
if (msg && msg.hasOwnProperty("error") && msg.error !== null) {
if (msg.error.hasOwnProperty("source") && msg.error.source !== null) {
@@ -253,268 +330,67 @@ class Flow {
}
}
}
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(throwingNode.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,
name: node.name,
count: count
}
};
if (logMessage.hasOwnProperty('stack')) {
errorMessage.error.stack = logMessage.stack;
}
targetCatchNode.receive(errorMessage);
handled = true;
});
this.catchNodes.forEach(function(targetCatchNode) {
if (targetCatchNode.scope && targetCatchNode.scope.indexOf(node.id) === -1) {
return;
}
if (!handled) {
throwingNode = this.activeNodes[throwingNode.z];
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,
name: node.name,
count: count
}
};
if (logMessage.hasOwnProperty('stack')) {
errorMessage.error.stack = logMessage.stack;
}
targetCatchNode.receive(errorMessage);
handled = true;
});
if (!handled) {
// Nothing in this flow handled the error - pass it to the parent
handled = this.parent.handleError(node,logMessage);
}
return handled;
}
dump() {
console.log("==================")
console.log(this.TYPE, this.id);
for (var id in this.activeNodes) {
if (this.activeNodes.hasOwnProperty(id)) {
var node = this.activeNodes[id];
console.log(" ",id.padEnd(16),node.type)
if (node.wires) {
console.log(" -> ",node.wires)
}
}
}
console.log("==================")
}
}
/**
* Create a new instance of a node
* @param {string} type The node type string
* @param {object} config The node configuration object
* @return {Node} The instance of the node
*/
function createNode(type,config) {
var newNode = null;
try {
var nodeTypeConstructor = typeRegistry.get(type);
if (nodeTypeConstructor) {
var conf = clone(config);
delete conf.credentials;
for (var p in conf) {
if (conf.hasOwnProperty(p)) {
flowUtil.mapEnvVarProperties(conf,p);
}
}
try {
newNode = new nodeTypeConstructor(conf);
} catch (err) {
Log.log({
level: Log.ERROR,
id:conf.id,
type: type,
msg: err
});
}
} else {
Log.error(Log._("nodes.flow.unknown-type", {type:type}));
}
} catch(err) {
Log.error(err);
}
return newNode;
}
function createSubflow(sf,sfn,subflows,globalSubflows,activeNodes) {
//console.log("CREATE SUBFLOW",sf.id,sfn.id);
var nodes = [];
var node_map = {};
var newNodes = [];
var node;
var wires;
var i,j,k;
var createNodeInSubflow = function(def) {
node = clone(def);
var nid = redUtil.generateId();
node_map[node.id] = node;
node._alias = node.id;
node.id = nid;
node.z = sfn.id;
newNodes.push(node);
}
// Clone all of the subflow config node definitions and give them new IDs
for (i in sf.configs) {
if (sf.configs.hasOwnProperty(i)) {
createNodeInSubflow(sf.configs[i]);
}
}
// Clone all of the subflow node definitions and give them new IDs
for (i in sf.nodes) {
if (sf.nodes.hasOwnProperty(i)) {
createNodeInSubflow(sf.nodes[i]);
}
}
// 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];
if (node.wires) {
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:""
})
} else {
for (var prop in node) {
if (node.hasOwnProperty(prop) && prop !== '_alias') {
if (node_map[node[prop]]) {
//console.log("Mapped",node.type,node.id,prop,node_map[node[prop]].id);
node[prop] = node_map[node[prop]].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.in) {
subflowInstance.wires = sf.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.out) {
var node,wires,i,j;
// Restore the original wiring to the internal nodes
subflowInstance.wires = clone(subflowInstance._originalWires);
for (i=0;i<sf.out.length;i++) {
wires = sf.out[i].wires;
for (j=0;j<wires.length;j++) {
if (wires[j].id != sf.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.out.length;i++) {
wires = sf.out[i].wires;
for (j=0;j<wires.length;j++) {
if (wires[j].id === sf.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.out) {
var modifiedNodes = {};
for (i=0;i<sf.out.length;i++) {
wires = sf.out[i].wires;
for (j=0;j<wires.length;j++) {
if (wires[j].id === sf.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) {
var newNode = createNode(type,node);
if (newNode) {
activeNodes[node.id] = newNode;
nodes.push(newNode);
}
} else {
var subflowId = m[1];
nodes = nodes.concat(createSubflow(subflows[subflowId]||globalSubflows[subflowId],node,subflows,globalSubflows,activeNodes));
}
}
subflowNode.instanceNodes = {};
nodes.forEach(function(node) {
subflowNode.instanceNodes[node.id] = node;
});
return nodes;
}
module.exports = {
init: function(runtime) {
nodeCloseTimeout = runtime.settings.nodeCloseTimeout || 15000;
Log = runtime.log;
Subflow = require("./Subflow");
Subflow.init(runtime);
},
create: function(global,conf) {
return new Flow(global,conf);
}
create: function(parent,global,conf) {
return new Flow(parent,global,conf);
},
Flow: Flow
}

246
packages/node_modules/@node-red/runtime/lib/nodes/flows/Subflow.js vendored Normal file
View File

@@ -0,0 +1,246 @@
/**
* Copyright JS Foundation and other contributors, http://js.foundation
*
* 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.
**/
const clone = require("clone");
const Flow = require('./Flow').Flow;
const redUtil = require("@node-red/util").util;
const flowUtil = require("./util");
var Log;
class Subflow extends Flow {
constructor(parent,globalFlow,subflowDef,subflowInstance) {
// console.log(subflowDef);
// console.log("CREATE SUBFLOW",subflowDef.id,subflowInstance.id);
// console.log("SubflowInstance\n"+JSON.stringify(subflowInstance," ",2));
// console.log("SubflowDef\n"+JSON.stringify(subflowDef," ",2));
var subflows = parent.flow.subflows;
var globalSubflows = parent.global.subflows;
var node_map = {};
var node;
var wires;
var i;
var subflowInternalFlowConfig = {
id: subflowInstance.id,
configs: {},
nodes: {},
subflows: {}
}
if (subflowDef.configs) {
// Clone all of the subflow config node definitions and give them new IDs
for (i in subflowDef.configs) {
if (subflowDef.configs.hasOwnProperty(i)) {
node = createNodeInSubflow(subflowInstance.id,subflowDef.configs[i]);
node_map[node._alias] = node;
subflowInternalFlowConfig.configs[node.id] = node;
}
}
}
if (subflowDef.nodes) {
// Clone all of the subflow node definitions and give them new IDs
for (i in subflowDef.nodes) {
if (subflowDef.nodes.hasOwnProperty(i)) {
node = createNodeInSubflow(subflowInstance.id,subflowDef.nodes[i]);
node_map[node._alias] = node;
subflowInternalFlowConfig.nodes[node.id] = node;
}
}
}
subflowInternalFlowConfig.subflows = clone(subflowDef.subflows || {});
remapSubflowNodes(subflowInternalFlowConfig.configs,node_map);
remapSubflowNodes(subflowInternalFlowConfig.nodes,node_map);
// console.log("Instance config\n",JSON.stringify(subflowInternalFlowConfig,"",2));
super(parent,globalFlow,subflowInternalFlowConfig);
this.TYPE = 'subflow';
this.subflowDef = subflowDef;
this.subflowInstance = subflowInstance;
this.node_map = node_map;
}
start(diff) {
var self = this;
// Create a subflow node to accept inbound messages and route appropriately
var Node = require("../Node");
var subflowInstanceConfig = {
id: this.subflowInstance.id,
type: this.subflowInstance.type,
z: this.subflowInstance.z,
name: this.subflowInstance.name,
wires: [],
_flow: this
}
if (this.subflowDef.in) {
subflowInstanceConfig.wires = this.subflowDef.in.map(function(n) { return n.wires.map(function(w) { return self.node_map[w.id].id;})})
subflowInstanceConfig._originalWires = clone(subflowInstanceConfig.wires);
}
this.node = new Node(subflowInstanceConfig);
this.node.on("input", function(msg) { this.send(msg);});
this.node._updateWires = this.node.updateWires;
this.node.updateWires = function(newWires) {
// Wire the subflow outputs
if (self.subflowDef.out) {
var node,wires,i,j;
// Restore the original wiring to the internal nodes
subflowInstanceConfig.wires = clone(subflowInstanceConfig._originalWires);
for (i=0;i<self.subflowDef.out.length;i++) {
wires = self.subflowDef.out[i].wires;
for (j=0;j<wires.length;j++) {
if (wires[j].id != self.subflowDef.id) {
node = self.node_map[wires[j].id];
if (node._originalWires) {
node.wires = clone(node._originalWires);
}
}
}
}
var modifiedNodes = {};
var subflowInstanceModified = false;
for (i=0;i<self.subflowDef.out.length;i++) {
wires = self.subflowDef.out[i].wires;
for (j=0;j<wires.length;j++) {
if (wires[j].id === self.subflowDef.id) {
subflowInstanceConfig.wires[wires[j].port] = subflowInstanceConfig.wires[wires[j].port].concat(newWires[i]);
subflowInstanceModified = true;
} else {
node = self.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];
self.activeNodes[id].updateWires(node.wires);
});
if (subflowInstanceModified) {
self.node._updateWires(subflowInstanceConfig.wires);
}
}
}
// Wire the subflow outputs
if (this.subflowDef.out) {
var modifiedNodes = {};
for (var i=0;i<this.subflowDef.out.length;i++) {
// i: the output index
// This is what this Output is wired to
wires = this.subflowDef.out[i].wires;
for (var j=0;j<wires.length;j++) {
if (wires[j].id === this.subflowDef.id) {
// A subflow input wired straight to a subflow output
subflowInstanceConfig.wires[wires[j].port] = subflowInstanceConfig.wires[wires[j].port].concat(this.subflowInstance.wires[i])
this.node._updateWires(subflowInstanceConfig.wires);
} else {
var node = self.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(this.subflowInstance.wires[i]);
}
}
}
}
super.start(diff);
}
stop(stopList,removedList) {
return super.stop(stopList,removedList);
}
}
function createNodeInSubflow(subflowInstanceId, def) {
let node = clone(def);
let nid = redUtil.generateId();
// console.log("Create Node In subflow",node.id, "--->",nid, "(",node.type,")")
// node_map[node.id] = node;
node._alias = node.id;
node.id = nid;
node.z = subflowInstanceId;
return node;
}
/**
* Given an object of {id:nodes} and a map of {old-id:node}, modifiy all
* properties in the nodes object to reference the new node ids.
* This handles:
* - node.wires,
* - node.scope of Catch and Status nodes,
* - node.XYZ for any property where XYZ is recognised as an old property
* @param {[type]} nodes [description]
* @param {[type]} nodeMap [description]
* @return {[type]} [description]
*/
function remapSubflowNodes(nodes,nodeMap) {
for (var id in nodes) {
if (nodes.hasOwnProperty(id)) {
var node = nodes[id];
if (node.wires) {
var outputs = node.wires;
for (j=0;j<outputs.length;j++) {
wires = outputs[j];
for (k=0;k<wires.length;k++) {
outputs[j][k] = nodeMap[outputs[j][k]].id
}
}
}
if ((node.type === 'catch' || node.type === 'status') && node.scope) {
node.scope = node.scope.map(function(id) {
return nodeMap[id]?nodeMap[id].id:""
})
} else {
for (var prop in node) {
if (node.hasOwnProperty(prop) && prop !== '_alias') {
if (nodeMap[node[prop]]) {
//console.log("Mapped",node.type,node.id,prop,nodeMap[node[prop]].id);
node[prop] = nodeMap[node[prop]].id;
}
}
}
}
}
}
}
function createSubflow(parent,globalFlow,subflowDef,subflowInstance) {
return new Subflow(parent,globalFlow,subflowDef,subflowInstance)
}
module.exports = {
init: function(runtime) {
Log = runtime.log;
},
create: createSubflow
}

98
packages/node_modules/@node-red/runtime/lib/nodes/flows/index.js vendored
View File

@@ -42,7 +42,6 @@ var started = false;
var credentialsPendingReset = false;
var activeNodesToFlow = {};
var subflowInstanceNodeMap = {};
var typeEventRegistered = false;
@@ -233,59 +232,6 @@ function getFlows() {
return activeConfig;
}
function delegateError(node,logMessage,msg) {
var handled = false;
if (activeFlows[node.z]) {
handled = activeFlows[node.z].handleError(node,logMessage,msg);
} else if (activeNodesToFlow[node.z] && activeFlows[activeNodesToFlow[node.z]]) {
handled = activeFlows[activeNodesToFlow[node.z]].handleError(node,logMessage,msg);
} else if (activeFlowConfig.subflows[node.z] && subflowInstanceNodeMap[node.id]) {
subflowInstanceNodeMap[node.id].forEach(function(n) {
handled = handled || delegateError(getNode(n),logMessage,msg);
});
}
return handled;
}
function handleError(node,logMessage,msg) {
var handled = false;
if (node.z) {
handled = delegateError(node,logMessage,msg);
} else {
if (activeFlowConfig.configs[node.id]) {
activeFlowConfig.configs[node.id]._users.forEach(function(id) {
var userNode = activeFlowConfig.allNodes[id];
handled = handled || delegateError(userNode,logMessage,msg);
})
}
}
return handled;
}
function delegateStatus(node,statusMessage) {
if (activeFlows[node.z]) {
activeFlows[node.z].handleStatus(node,statusMessage);
} else if (activeNodesToFlow[node.z] && activeFlows[activeNodesToFlow[node.z]]) {
activeFlows[activeNodesToFlow[node.z]].handleStatus(node,statusMessage);
}
}
function handleStatus(node,statusMessage) {
events.emit("node-status",{
id: node.id,
status:statusMessage
});
if (node.z) {
delegateStatus(node,statusMessage);
} else {
if (activeFlowConfig.configs[node.id]) {
activeFlowConfig.configs[node.id]._users.forEach(function(id) {
var userNode = activeFlowConfig.allNodes[id];
delegateStatus(userNode,statusMessage);
})
}
}
}
function start(type,diff,muteLog) {
type = type||"full";
started = true;
@@ -338,7 +284,7 @@ function start(type,diff,muteLog) {
// Check the 'global' flow is running
if (!activeFlows['global']) {
log.debug("red/nodes/flows.start : starting flow : global");
activeFlows['global'] = Flow.create(activeFlowConfig);
activeFlows['global'] = Flow.create(flowAPI,activeFlowConfig);
}
// Check each flow in the active configuration
@@ -346,7 +292,7 @@ function start(type,diff,muteLog) {
if (activeFlowConfig.flows.hasOwnProperty(id)) {
if (!activeFlowConfig.flows[id].disabled && !activeFlows[id]) {
// This flow is not disabled, nor is it currently active, so create it
activeFlows[id] = Flow.create(activeFlowConfig,activeFlowConfig.flows[id]);
activeFlows[id] = Flow.create(flowAPI,activeFlowConfig,activeFlowConfig.flows[id]);
log.debug("red/nodes/flows.start : starting flow : "+id);
} else {
log.debug("red/nodes/flows.start : not starting disabled flow : "+id);
@@ -366,7 +312,7 @@ function start(type,diff,muteLog) {
activeFlows[id].update(activeFlowConfig,activeFlowConfig.flows[id]);
} else {
// This flow didn't previously exist, so create it
activeFlows[id] = Flow.create(activeFlowConfig,activeFlowConfig.flows[id]);
activeFlows[id] = Flow.create(flowAPI,activeFlowConfig,activeFlowConfig.flows[id]);
log.debug("red/nodes/flows.start : starting flow : "+id);
}
} else {
@@ -375,22 +321,20 @@ function start(type,diff,muteLog) {
}
}
}
// Having created or updated all flows, now start them.
for (id in activeFlows) {
if (activeFlows.hasOwnProperty(id)) {
activeFlows[id].start(diff);
// Create a map of node id to flow id and also a subflowInstance lookup map
var activeNodes = activeFlows[id].getActiveNodes();
Object.keys(activeNodes).forEach(function(nid) {
activeNodesToFlow[nid] = id;
if (activeNodes[nid]._alias) {
subflowInstanceNodeMap[activeNodes[nid]._alias] = subflowInstanceNodeMap[activeNodes[nid]._alias] || [];
subflowInstanceNodeMap[activeNodes[nid]._alias].push(nid);
}
});
try {
activeFlows[id].start(diff);
// Create a map of node id to flow id and also a subflowInstance lookup map
var activeNodes = activeFlows[id].getActiveNodes();
Object.keys(activeNodes).forEach(function(nid) {
activeNodesToFlow[nid] = id;
});
} catch(err) {
console.log(err.stack);
}
}
}
events.emit("nodes-started");
@@ -465,11 +409,6 @@ function stop(type,diff,muteLog) {
delete activeNodesToFlow[id];
});
}
// Ideally we'd prune just what got stopped - but mapping stopList
// id to the list of subflow instance nodes is something only Flow
// can do... so cheat by wiping the map knowing it'll be rebuilt
// in start()
subflowInstanceNodeMap = {};
if (!muteLog) {
if (type !== "full") {
log.info(log._("nodes.flows.stopped-modified-"+type));
@@ -735,6 +674,13 @@ function removeFlow(id) {
});
}
const flowAPI = {
getNode: getNode,
handleError: () => false,
handleStatus: () => false
}
module.exports = {
init: init,
@@ -773,8 +719,8 @@ module.exports = {
get started() { return started },
handleError: handleError,
handleStatus: handleStatus,
// handleError: handleError,
// handleStatus: handleStatus,
checkTypeInUse: checkTypeInUse,

47
packages/node_modules/@node-red/runtime/lib/nodes/flows/util.js vendored
View File

@@ -15,6 +15,7 @@
**/
var clone = require("clone");
var redUtil = require("@node-red/util").util;
var Log = require("@node-red/util").log;
var subflowInstanceRE = /^subflow:(.+)$/;
var typeRegistry = require("@node-red/registry");
@@ -427,9 +428,10 @@ module.exports = {
// console.log(diff);
// for (id in newConfig.allNodes) {
// console.log(
// (added[id]?"+":(changed[id]?"!":" "))+(wiringChanged[id]?"w":" ")+(diff.linked.indexOf(id)!==-1?"~":" "),
// id,
// newConfig.allNodes[id].type,
// (added[id]?"a":(changed[id]?"c":" "))+(wiringChanged[id]?"w":" ")+(diff.linked.indexOf(id)!==-1?"l":" "),
// newConfig.allNodes[id].type.padEnd(10),
// id.padEnd(16),
// (newConfig.allNodes[id].z||"").padEnd(16),
// newConfig.allNodes[id].name||newConfig.allNodes[id].label||""
// );
// }
@@ -443,5 +445,44 @@ module.exports = {
// }
return diff;
},
/**
* Create a new instance of a node
* @param {Flow} flow The containing flow
* @param {object} config The node configuration object
* @return {Node} The instance of the node
*/
createNode: function(flow,config) {
var newNode = null;
var type = config.type;
try {
var nodeTypeConstructor = typeRegistry.get(type);
if (nodeTypeConstructor) {
var conf = clone(config);
delete conf.credentials;
for (var p in conf) {
if (conf.hasOwnProperty(p)) {
mapEnvVarProperties(conf,p);
}
}
try {
conf._flow = flow;
newNode = new nodeTypeConstructor(conf);
} catch (err) {
Log.log({
level: Log.ERROR,
id:conf.id,
type: type,
msg: err
});
}
} else {
Log.error(Log._("nodes.flow.unknown-type", {type:type}));
}
} catch(err) {
Log.error(err);
}
return newNode;
}
}