node-red
This commit is contained in:
gongwenxin
@@ -0,0 +1,857 @@
|
||||
/**
|
||||
* 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 redUtil = require("@node-red/util").util;
|
||||
const events = require("@node-red/util").events;
|
||||
const flowUtil = require("./util");
|
||||
const context = require('../nodes/context');
|
||||
const hooks = require("@node-red/util").hooks;
|
||||
const credentials = require("../nodes/credentials");
|
||||
|
||||
let Subflow;
|
||||
let Log;
|
||||
let Group;
|
||||
|
||||
let nodeCloseTimeout = 15000;
|
||||
let asyncMessageDelivery = true;
|
||||
|
||||
/**
|
||||
* This class represents a flow within the runtime. It is responsible for
|
||||
* creating, starting and stopping all nodes within the flow.
|
||||
*/
|
||||
class Flow {
|
||||
|
||||
/**
|
||||
* Create a Flow object.
|
||||
* @param {[type]} parent The parent flow
|
||||
* @param {[type]} globalFlow The global flow definition
|
||||
* @param {[type]} flow This flow's definition
|
||||
*/
|
||||
constructor(parent,globalFlow,flow) {
|
||||
this.TYPE = 'flow';
|
||||
this.parent = parent;
|
||||
this.global = globalFlow;
|
||||
if (typeof flow === 'undefined') {
|
||||
this.flow = globalFlow;
|
||||
this.isGlobalFlow = true;
|
||||
} else {
|
||||
this.flow = flow;
|
||||
this.isGlobalFlow = false;
|
||||
}
|
||||
this.id = this.flow.id || "global";
|
||||
this.groups = {}
|
||||
this.groupOrder = []
|
||||
this.activeNodes = {};
|
||||
this.subflowInstanceNodes = {};
|
||||
this.catchNodes = [];
|
||||
this.statusNodes = [];
|
||||
this.path = this.id;
|
||||
// Ensure a context exists for this flow
|
||||
this.context = context.getFlowContext(this.id,this.parent.id);
|
||||
|
||||
// env is an array of env definitions
|
||||
// _env is an object for direct lookup of env name -> value
|
||||
this.env = this.flow.env
|
||||
this._env = {}
|
||||
}
|
||||
|
||||
/**
|
||||
* Log a debug-level message from this flow
|
||||
* @param {[type]} msg [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
debug(msg) {
|
||||
Log.log({
|
||||
id: this.id||"global",
|
||||
level: Log.DEBUG,
|
||||
type:this.TYPE,
|
||||
msg:msg
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Log an error-level message from this flow
|
||||
* @param {[type]} msg [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
error(msg) {
|
||||
Log.log({
|
||||
id: this.id||"global",
|
||||
level: Log.ERROR,
|
||||
type:this.TYPE,
|
||||
msg:msg
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Log a info-level message from this flow
|
||||
* @param {[type]} msg [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
info(msg) {
|
||||
Log.log({
|
||||
id: this.id||"global",
|
||||
level: Log.INFO,
|
||||
type:this.TYPE,
|
||||
msg:msg
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Log a trace-level message from this flow
|
||||
* @param {[type]} msg [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
trace(msg) {
|
||||
Log.log({
|
||||
id: this.id||"global",
|
||||
level: Log.TRACE,
|
||||
type:this.TYPE,
|
||||
msg:msg
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* [log description]
|
||||
* @param {[type]} msg [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
log(msg) {
|
||||
if (!msg.path) {
|
||||
msg.path = this.path;
|
||||
}
|
||||
this.parent.log(msg);
|
||||
}
|
||||
|
||||
/**
|
||||
* Start this flow.
|
||||
* The `diff` argument helps define what needs to be started in the case
|
||||
* of a modified-nodes/flows type deploy.
|
||||
* @param {[type]} msg [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
async start(diff) {
|
||||
this.trace("start "+this.TYPE+" ["+this.path+"]");
|
||||
var node;
|
||||
var newNode;
|
||||
var id;
|
||||
this.catchNodes = [];
|
||||
this.statusNodes = [];
|
||||
this.completeNodeMap = {};
|
||||
|
||||
|
||||
if (this.isGlobalFlow) {
|
||||
// This is the global flow. It needs to go find the `global-config`
|
||||
// node and extract any env properties from it
|
||||
const configNodes = Object.keys(this.flow.configs);
|
||||
for (let i = 0; i < configNodes.length; i++) {
|
||||
const node = this.flow.configs[configNodes[i]]
|
||||
if (node.type === 'global-config' && node.env) {
|
||||
const globalCreds = credentials.get(node.id)?.map || {}
|
||||
const nodeEnv = await flowUtil.evaluateEnvProperties(this, node.env, globalCreds)
|
||||
this._env = { ...this._env, ...nodeEnv }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (this.env) {
|
||||
this._env = { ...this._env, ...await flowUtil.evaluateEnvProperties(this, this.env, credentials.get(this.id)) }
|
||||
}
|
||||
|
||||
// Initialise the group objects. These must be done in the right order
|
||||
// starting from outer-most to inner-most so that the parent hierarchy
|
||||
// is maintained.
|
||||
this.groups = {}
|
||||
this.groupOrder = []
|
||||
const groupIds = Object.keys(this.flow.groups || {})
|
||||
while (groupIds.length > 0) {
|
||||
const id = groupIds.shift()
|
||||
const groupDef = this.flow.groups[id]
|
||||
if (!groupDef.g || this.groups[groupDef.g]) {
|
||||
// The parent of this group is available - either another group
|
||||
// or the top-level flow (this)
|
||||
const parent = this.groups[groupDef.g] || this
|
||||
this.groups[groupDef.id] = new Group(parent, groupDef)
|
||||
this.groupOrder.push(groupDef.id)
|
||||
} else {
|
||||
// Try again once we've processed the other groups
|
||||
groupIds.push(id)
|
||||
}
|
||||
}
|
||||
|
||||
for (let i = 0; i < this.groupOrder.length; i++) {
|
||||
// Start the groups in the right order so they
|
||||
// can setup their env vars knowning their parent
|
||||
// will have been started
|
||||
await this.groups[this.groupOrder[i]].start()
|
||||
}
|
||||
|
||||
var configNodes = Object.keys(this.flow.configs);
|
||||
var configNodeAttempts = {};
|
||||
while (configNodes.length > 0) {
|
||||
id = configNodes.shift();
|
||||
node = this.flow.configs[id];
|
||||
if (!this.activeNodes[id]) {
|
||||
if (node.d !== true) {
|
||||
var readyToCreate = true;
|
||||
// This node doesn't exist.
|
||||
// Check it doesn't reference another non-existent config node
|
||||
for (var prop in node) {
|
||||
if (node.hasOwnProperty(prop) &&
|
||||
prop !== 'id' &&
|
||||
prop !== 'wires' &&
|
||||
prop !== '_users' &&
|
||||
this.flow.configs[node[prop]] &&
|
||||
this.flow.configs[node[prop]].d !== true
|
||||
) {
|
||||
if (!this.activeNodes[node[prop]]) {
|
||||
// References a non-existent config node
|
||||
// Add it to the back of the list to try again later
|
||||
configNodes.push(id);
|
||||
configNodeAttempts[id] = (configNodeAttempts[id]||0)+1;
|
||||
if (configNodeAttempts[id] === 100) {
|
||||
throw new Error("Circular config node dependency detected: "+id);
|
||||
}
|
||||
readyToCreate = false;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (readyToCreate) {
|
||||
newNode = await flowUtil.createNode(this,node);
|
||||
if (newNode) {
|
||||
this.activeNodes[id] = newNode;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
this.debug("not starting disabled config node : "+id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (diff && diff.rewired) {
|
||||
for (var j=0;j<diff.rewired.length;j++) {
|
||||
var rewireNode = this.activeNodes[diff.rewired[j]];
|
||||
if (rewireNode) {
|
||||
rewireNode.updateWires(this.flow.nodes[rewireNode.id].wires);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (id in this.flow.nodes) {
|
||||
if (this.flow.nodes.hasOwnProperty(id)) {
|
||||
node = this.flow.nodes[id];
|
||||
if (node.d !== true) {
|
||||
if (!node.subflow) {
|
||||
if (!this.activeNodes[id]) {
|
||||
newNode = await flowUtil.createNode(this,node);
|
||||
if (newNode) {
|
||||
this.activeNodes[id] = newNode;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
if (!this.subflowInstanceNodes[id]) {
|
||||
try {
|
||||
var subflowDefinition = this.flow.subflows[node.subflow]||this.global.subflows[node.subflow]
|
||||
// console.log("NEED TO CREATE A SUBFLOW",id,node.subflow);
|
||||
this.subflowInstanceNodes[id] = true;
|
||||
var subflow = Subflow.create(
|
||||
this,
|
||||
this.global,
|
||||
subflowDefinition,
|
||||
node
|
||||
);
|
||||
this.subflowInstanceNodes[id] = subflow;
|
||||
await subflow.start();
|
||||
this.activeNodes[id] = subflow.node;
|
||||
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
this.debug("not starting disabled node : "+id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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/complete 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||"")+(node._zAlias?" [zAlias:"+node._zAlias+"]":""));
|
||||
if (node.type === "catch") {
|
||||
this.catchNodes.push(node);
|
||||
} else if (node.type === "status") {
|
||||
this.statusNodes.push(node);
|
||||
} else if (node.type === "complete") {
|
||||
if (node.scope) {
|
||||
node.scope.forEach(id => {
|
||||
this.completeNodeMap[id] = this.completeNodeMap[id] || [];
|
||||
this.completeNodeMap[id].push(node);
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
this.catchNodes.sort(function(A,B) {
|
||||
if (A.scope && !B.scope) {
|
||||
return -1;
|
||||
} else if (!A.scope && B.scope) {
|
||||
return 1;
|
||||
} else if (A.scope && B.scope) {
|
||||
return 0;
|
||||
} else if (A.uncaught && !B.uncaught) {
|
||||
return 1;
|
||||
} else if (!A.uncaught && B.uncaught) {
|
||||
return -1;
|
||||
}
|
||||
return 0;
|
||||
});
|
||||
|
||||
if (activeCount > 0) {
|
||||
this.trace("------------------|--------------|-----------------");
|
||||
}
|
||||
// this.dump();
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop this flow.
|
||||
* The `stopList` argument helps define what needs to be stopped in the case
|
||||
* of a modified-nodes/flows type deploy.
|
||||
* @param {[type]} stopList [description]
|
||||
* @param {[type]} removedList [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
stop(stopList, removedList) {
|
||||
this.trace("stop "+this.TYPE);
|
||||
var i;
|
||||
if (!stopList) {
|
||||
stopList = Object.keys(this.activeNodes);
|
||||
}
|
||||
// this.trace(" stopList: "+stopList.join(","))
|
||||
// Convert the list to a map to avoid multiple scans of the list
|
||||
var removedMap = {};
|
||||
removedList = removedList || [];
|
||||
removedList.forEach(function(id) {
|
||||
removedMap[id] = true;
|
||||
});
|
||||
|
||||
let nodesToStop = [];
|
||||
let configsToStop = [];
|
||||
stopList.forEach(id => {
|
||||
if (this.flow.configs[id]) {
|
||||
configsToStop.push(id);
|
||||
} else {
|
||||
nodesToStop.push(id);
|
||||
}
|
||||
});
|
||||
stopList = nodesToStop.concat(configsToStop);
|
||||
|
||||
var promises = [];
|
||||
for (i=0;i<stopList.length;i++) {
|
||||
var node = this.activeNodes[stopList[i]];
|
||||
if (node) {
|
||||
delete this.activeNodes[stopList[i]];
|
||||
if (this.subflowInstanceNodes[stopList[i]]) {
|
||||
delete this.subflowInstanceNodes[stopList[i]];
|
||||
}
|
||||
try {
|
||||
var removed = removedMap[stopList[i]];
|
||||
promises.push(stopNode(node,removed).catch(()=>{}));
|
||||
} catch(err) {
|
||||
node.error(err);
|
||||
}
|
||||
if (removedMap[stopList[i]]) {
|
||||
events.emit("node-status",{
|
||||
id: node.id
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
return Promise.all(promises);
|
||||
}
|
||||
|
||||
/**
|
||||
* Update the flow definition. This doesn't change anything that is running.
|
||||
* This should be called after `stop` and before `start`.
|
||||
* @param {[type]} _global [description]
|
||||
* @param {[type]} _flow [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
update(_global,_flow) {
|
||||
this.global = _global;
|
||||
this.flow = _flow;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get a node instance from this flow. If the node is not known to this
|
||||
* flow, pass the request up to the parent.
|
||||
* @param {String} id [description]
|
||||
* @param {Boolean} cancelBubble if true, prevents the flow from passing the request to the parent
|
||||
* This stops infinite loops when the parent asked this Flow for the
|
||||
* node to begin with.
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
getNode(id, cancelBubble) {
|
||||
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.flow.nodes[id].type.substring(0,8) != "subflow:")) {
|
||||
// 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 or subflow instance node
|
||||
return this.activeNodes[id];
|
||||
} else if (this.subflowInstanceNodes[id]) {
|
||||
return this.subflowInstanceNodes[id];
|
||||
} else if (cancelBubble) {
|
||||
// The node could be inside one of this flow's subflows
|
||||
var node;
|
||||
for (var sfId in this.subflowInstanceNodes) {
|
||||
if (this.subflowInstanceNodes.hasOwnProperty(sfId)) {
|
||||
node = this.subflowInstanceNodes[sfId].getNode(id,cancelBubble);
|
||||
if (node) {
|
||||
return node;
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Node not found inside this flow - ask the parent
|
||||
return this.parent.getNode(id);
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Get a group node instance
|
||||
* @param {String} id
|
||||
* @return {Node} group node
|
||||
*/
|
||||
getGroupNode(id) {
|
||||
return this.groups[id];
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all of the nodes instantiated within this flow
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
getActiveNodes() {
|
||||
return this.activeNodes;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get a flow setting value.
|
||||
* @param {[type]} key [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
getSetting(key) {
|
||||
const flow = this.flow;
|
||||
if (key === "NR_FLOW_NAME") {
|
||||
return flow.label;
|
||||
}
|
||||
if (key === "NR_FLOW_ID") {
|
||||
return flow.id;
|
||||
}
|
||||
if (!key.startsWith("$parent.")) {
|
||||
if (this._env.hasOwnProperty(key)) {
|
||||
return (this._env[key] && Object.hasOwn(this._env[key], 'value') && this._env[key].__clone__) ? clone(this._env[key].value) : this._env[key]
|
||||
}
|
||||
} else {
|
||||
key = key.substring(8);
|
||||
}
|
||||
// Delegate to the parent flow.
|
||||
return this.parent.getSetting(key);
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle a status event from a node within this flow.
|
||||
* @param {Node} node The original node that triggered the event
|
||||
* @param {Object} statusMessage The status object
|
||||
* @param {Node} reportingNode The node emitting the status event.
|
||||
* This could be a subflow instance node when the status
|
||||
* is being delegated up.
|
||||
* @param {boolean} muteStatusEvent Whether to emit the status event
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
handleStatus(node,statusMessage,reportingNode,muteStatusEvent) {
|
||||
if (!reportingNode) {
|
||||
reportingNode = node;
|
||||
}
|
||||
if (!muteStatusEvent) {
|
||||
if (statusMessage.hasOwnProperty("text") && typeof statusMessage.text !== "string") {
|
||||
try {
|
||||
statusMessage.text = statusMessage.text.toString();
|
||||
}
|
||||
catch(e) {}
|
||||
}
|
||||
events.emit("node-status",{
|
||||
id: node.id,
|
||||
status:statusMessage
|
||||
});
|
||||
}
|
||||
|
||||
let handled = false;
|
||||
|
||||
if (this.id === 'global' && node.users) {
|
||||
// This is a global config node
|
||||
// Delegate status to any nodes using this config node
|
||||
for (let userNode in node.users) {
|
||||
if (node.users.hasOwnProperty(userNode)) {
|
||||
handled = node.users[userNode]._flow.handleStatus(node,statusMessage,node.users[userNode],true) || handled;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
const candidateNodes = [];
|
||||
this.statusNodes.forEach(targetStatusNode => {
|
||||
if (targetStatusNode.g && targetStatusNode.scope === 'group' && !reportingNode.g) {
|
||||
// Status node inside a group, reporting node not in a group - skip it
|
||||
return
|
||||
}
|
||||
if (Array.isArray(targetStatusNode.scope) && targetStatusNode.scope.indexOf(reportingNode.id) === -1) {
|
||||
return;
|
||||
}
|
||||
let distance = 0
|
||||
if (reportingNode.g) {
|
||||
// Reporting node inside a group. Calculate the distance between it and the status node
|
||||
let containingGroup = this.groups[reportingNode.g]
|
||||
while (containingGroup && containingGroup.id !== targetStatusNode.g) {
|
||||
distance++
|
||||
containingGroup = this.groups[containingGroup.g]
|
||||
}
|
||||
if (!containingGroup && targetStatusNode.g && targetStatusNode.scope === 'group') {
|
||||
// This status node is in a group, but not in the same hierachy
|
||||
// the reporting node is in
|
||||
return
|
||||
}
|
||||
}
|
||||
candidateNodes.push({ d: distance, n: targetStatusNode })
|
||||
})
|
||||
candidateNodes.sort((A,B) => {
|
||||
return A.d - B.d
|
||||
})
|
||||
candidateNodes.forEach(candidate => {
|
||||
const targetStatusNode = candidate.n
|
||||
var message = {
|
||||
status: clone(statusMessage)
|
||||
}
|
||||
if (statusMessage.hasOwnProperty("text")) {
|
||||
message.status.text = statusMessage.text.toString();
|
||||
}
|
||||
message.status.source = {
|
||||
id: node.id,
|
||||
type: node.type,
|
||||
name: node.name
|
||||
}
|
||||
|
||||
targetStatusNode.receive(message);
|
||||
handled = true;
|
||||
});
|
||||
}
|
||||
return handled;
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle an error event from a node within this flow. If there are no Catch
|
||||
* nodes within this flow, pass the event to the parent flow.
|
||||
* @param {[type]} node [description]
|
||||
* @param {[type]} logMessage [description]
|
||||
* @param {[type]} msg [description]
|
||||
* @param {[type]} reportingNode [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
handleError(node,logMessage,msg,reportingNode) {
|
||||
if (!reportingNode) {
|
||||
reportingNode = node;
|
||||
}
|
||||
// console.log("HE",logMessage);
|
||||
var count = 1;
|
||||
if (msg && msg.hasOwnProperty("error") && msg.error) {
|
||||
if (msg.error.hasOwnProperty("source") && msg.error.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 false;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
let handled = false;
|
||||
|
||||
if (this.id === 'global' && node.users) {
|
||||
// This is a global config node
|
||||
// Delegate status to any nodes using this config node
|
||||
for (let userNode in node.users) {
|
||||
if (node.users.hasOwnProperty(userNode)) {
|
||||
handled = node.users[userNode]._flow.handleError(node,logMessage,msg,node.users[userNode]) || handled;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
const candidateNodes = [];
|
||||
this.catchNodes.forEach(targetCatchNode => {
|
||||
if (targetCatchNode.g && targetCatchNode.scope === 'group' && !reportingNode.g) {
|
||||
// Catch node inside a group, reporting node not in a group - skip it
|
||||
return
|
||||
}
|
||||
if (Array.isArray(targetCatchNode.scope) && targetCatchNode.scope.indexOf(reportingNode.id) === -1) {
|
||||
// Catch node has a scope set and it doesn't include the reporting node
|
||||
return;
|
||||
}
|
||||
let distance = 0
|
||||
if (reportingNode.g) {
|
||||
// Reporting node inside a group. Calculate the distance between it and the catch node
|
||||
let containingGroup = this.groups[reportingNode.g]
|
||||
while (containingGroup && containingGroup.id !== targetCatchNode.g) {
|
||||
distance++
|
||||
containingGroup = this.groups[containingGroup.g]
|
||||
}
|
||||
if (!containingGroup && targetCatchNode.g && targetCatchNode.scope === 'group') {
|
||||
// This catch node is in a group, but not in the same hierachy
|
||||
// the reporting node is in
|
||||
return
|
||||
}
|
||||
}
|
||||
candidateNodes.push({ d: distance, n: targetCatchNode })
|
||||
})
|
||||
candidateNodes.sort((A,B) => {
|
||||
return A.d - B.d
|
||||
})
|
||||
let handledByUncaught = false
|
||||
candidateNodes.forEach(candidate => {
|
||||
const targetCatchNode = candidate.n
|
||||
if (targetCatchNode.uncaught && !handledByUncaught) {
|
||||
// This node only wants errors that haven't already been handled
|
||||
if (handled) {
|
||||
return
|
||||
}
|
||||
handledByUncaught = true
|
||||
}
|
||||
let 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('code')) {
|
||||
errorMessage.error.code = logMessage.code;
|
||||
}
|
||||
if (logMessage.hasOwnProperty('stack')) {
|
||||
errorMessage.error.stack = logMessage.stack;
|
||||
}
|
||||
if (logMessage.hasOwnProperty('cause')) {
|
||||
errorMessage.error.cause = logMessage.cause;
|
||||
}
|
||||
targetCatchNode.receive(errorMessage);
|
||||
handled = true;
|
||||
});
|
||||
}
|
||||
return handled;
|
||||
}
|
||||
|
||||
handleComplete(node,msg) {
|
||||
if (this.completeNodeMap[node.id]) {
|
||||
let toSend = msg;
|
||||
this.completeNodeMap[node.id].forEach((completeNode,index) => {
|
||||
toSend = redUtil.cloneMessage(msg);
|
||||
completeNode.receive(toSend);
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
send(sendEvents) {
|
||||
// onSend - passed an array of SendEvent objects. The messages inside these objects are exactly what the node has passed to node.send - meaning there could be duplicate references to the same message object.
|
||||
// preRoute - called once for each SendEvent object in turn
|
||||
// preDeliver - the local router has identified the node it is going to send to. At this point, the message has been cloned if needed.
|
||||
// postDeliver - the message has been dispatched to be delivered asynchronously (unless the sync delivery flag is set, in which case it would be continue as synchronous delivery)
|
||||
// onReceive - a node is about to receive a message
|
||||
// postReceive - the message has been passed to the node's input handler
|
||||
// onDone, onError - the node has completed with a message or logged an error
|
||||
handleOnSend(this,sendEvents, (err, eventData) => {
|
||||
if (err) {
|
||||
let srcNode;
|
||||
if (Array.isArray(eventData)) {
|
||||
srcNode = eventData[0].source.node;
|
||||
} else {
|
||||
srcNode = eventData.source.node;
|
||||
}
|
||||
srcNode.error(err);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
getContext(scope) {
|
||||
if (scope === 'flow') {
|
||||
return this.context
|
||||
} else if (scope === 'global') {
|
||||
return context.get('global')
|
||||
}
|
||||
}
|
||||
|
||||
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("==================")
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop an individual node within this flow.
|
||||
*
|
||||
* @param {[type]} node [description]
|
||||
* @param {[type]} removed [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
function stopNode(node,removed) {
|
||||
Log.trace("Stopping node "+node.type+":"+node.id+(removed?" removed":""));
|
||||
const start = Date.now();
|
||||
const closePromise = node.close(removed);
|
||||
let closeTimer = null;
|
||||
const closeTimeout = new Promise((resolve,reject) => {
|
||||
closeTimer = setTimeout(() => {
|
||||
reject("Close timed out");
|
||||
}, nodeCloseTimeout);
|
||||
});
|
||||
return Promise.race([closePromise,closeTimeout]).then(() => {
|
||||
clearTimeout(closeTimer);
|
||||
var delta = Date.now() - start;
|
||||
Log.trace("Stopped node "+node.type+":"+node.id+" ("+delta+"ms)" );
|
||||
}).catch(err => {
|
||||
clearTimeout(closeTimer);
|
||||
node.error(Log._("nodes.flows.stopping-error",{message:err}));
|
||||
Log.debug(err.stack);
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
function handleOnSend(flow,sendEvents, reportError) {
|
||||
// onSend - passed an array of SendEvent objects. The messages inside these objects are exactly what the node has passed to node.send - meaning there could be duplicate references to the same message object.
|
||||
hooks.trigger("onSend",sendEvents,(err) => {
|
||||
if (err) {
|
||||
reportError(err,sendEvents);
|
||||
return
|
||||
} else if (err !== false) {
|
||||
for (var i=0;i<sendEvents.length;i++) {
|
||||
handlePreRoute(flow,sendEvents[i],reportError)
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function handlePreRoute(flow, sendEvent, reportError) {
|
||||
// preRoute - called once for each SendEvent object in turn
|
||||
hooks.trigger("preRoute",sendEvent,(err) => {
|
||||
if (err) {
|
||||
reportError(err,sendEvent);
|
||||
return;
|
||||
} else if (err !== false) {
|
||||
sendEvent.destination.node = flow.getNode(sendEvent.destination.id);
|
||||
if (sendEvent.destination.node && typeof sendEvent.destination.node === 'object') {
|
||||
if (sendEvent.cloneMessage) {
|
||||
sendEvent.msg = redUtil.cloneMessage(sendEvent.msg);
|
||||
}
|
||||
handlePreDeliver(flow,sendEvent,reportError);
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
function deliverMessageToDestination(sendEvent) {
|
||||
if (sendEvent?.destination?.node) {
|
||||
try {
|
||||
sendEvent.destination.node.receive(sendEvent.msg);
|
||||
} catch(err) {
|
||||
Log.error(`Error delivering message to node:${sendEvent.destination.node._path} [${sendEvent.destination.node.type}]`)
|
||||
Log.error(err.stack)
|
||||
}
|
||||
}
|
||||
}
|
||||
function handlePreDeliver(flow,sendEvent, reportError) {
|
||||
// preDeliver - the local router has identified the node it is going to send to. At this point, the message has been cloned if needed.
|
||||
hooks.trigger("preDeliver",sendEvent,(err) => {
|
||||
if (err) {
|
||||
reportError(err,sendEvent);
|
||||
return;
|
||||
} else if (err !== false) {
|
||||
if (asyncMessageDelivery) {
|
||||
setImmediate(function() {
|
||||
deliverMessageToDestination(sendEvent)
|
||||
})
|
||||
} else {
|
||||
deliverMessageToDestination(sendEvent)
|
||||
}
|
||||
// postDeliver - the message has been dispatched to be delivered asynchronously (unless the sync delivery flag is set, in which case it would be continue as synchronous delivery)
|
||||
hooks.trigger("postDeliver", sendEvent, function(err) {
|
||||
if (err) {
|
||||
reportError(err,sendEvent);
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
init: function(runtime) {
|
||||
nodeCloseTimeout = runtime.settings.nodeCloseTimeout || 15000;
|
||||
asyncMessageDelivery = !runtime.settings.runtimeSyncDelivery
|
||||
Log = runtime.log;
|
||||
Subflow = require("./Subflow");
|
||||
Group = require("./Group").Group
|
||||
},
|
||||
create: function(parent,global,conf) {
|
||||
return new Flow(parent,global,conf)
|
||||
},
|
||||
Flow: Flow
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
const flowUtil = require("./util");
|
||||
const credentials = require("../nodes/credentials");
|
||||
const clone = require("clone");
|
||||
|
||||
/**
|
||||
* This class represents a group within the runtime.
|
||||
*/
|
||||
class Group {
|
||||
|
||||
/**
|
||||
* Create a Group object.
|
||||
* @param {[type]} parent The parent flow/group
|
||||
* @param {[type]} groupDef This group's definition
|
||||
*/
|
||||
constructor(parent, groupDef) {
|
||||
this.TYPE = 'group'
|
||||
this.name = groupDef.name
|
||||
this.parent = parent
|
||||
this.group = groupDef
|
||||
this.id = this.group.id
|
||||
this.g = this.group.g
|
||||
this.env = this.group.env
|
||||
this._env = {}
|
||||
}
|
||||
|
||||
async start() {
|
||||
if (this.env) {
|
||||
this._env = await flowUtil.evaluateEnvProperties(this, this.env, credentials.get(this.id))
|
||||
}
|
||||
}
|
||||
/**
|
||||
* Get a group setting value.
|
||||
* @param {[type]} key [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
getSetting(key) {
|
||||
if (key === "NR_GROUP_NAME") {
|
||||
return this.name;
|
||||
}
|
||||
if (key === "NR_GROUP_ID") {
|
||||
return this.id;
|
||||
}
|
||||
if (!key.startsWith("$parent.")) {
|
||||
if (this._env.hasOwnProperty(key)) {
|
||||
return (this._env[key] && Object.hasOwn(this._env[key], 'value') && this._env[key].__clone__) ? clone(this._env[key].value) : this._env[key]
|
||||
}
|
||||
} else {
|
||||
key = key.substring(8);
|
||||
}
|
||||
return this.parent.getSetting(key);
|
||||
}
|
||||
|
||||
error(msg) {
|
||||
this.parent.error(msg);
|
||||
}
|
||||
|
||||
getContext(scope) {
|
||||
return this.parent.getContext(scope);
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
Group
|
||||
}
|
||||
@@ -0,0 +1,568 @@
|
||||
/**
|
||||
* 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 jsonClone = require("rfdc")();
|
||||
const Flow = require('./Flow').Flow;
|
||||
const context = require('../nodes/context');
|
||||
const util = require("util");
|
||||
const redUtil = require("@node-red/util").util;
|
||||
const events = require("@node-red/util").events;
|
||||
const flowUtil = require("./util");
|
||||
const credentials = require("../nodes/credentials");
|
||||
|
||||
/**
|
||||
* Create deep copy of object
|
||||
*/
|
||||
function deepCopy(obj) {
|
||||
return JSON.parse(JSON.stringify(obj));
|
||||
}
|
||||
|
||||
/**
|
||||
* Evaluate Input Value
|
||||
*/
|
||||
function evaluateInputValue(value, type, node) {
|
||||
if (type === "bool") {
|
||||
return (value === "true") || (value === true);
|
||||
}
|
||||
if (type === "cred") {
|
||||
return value;
|
||||
}
|
||||
return redUtil.evaluateNodeProperty(value, type, node, null, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* This class represents a subflow - which is handled as a special type of Flow
|
||||
*/
|
||||
class Subflow extends Flow {
|
||||
|
||||
/**
|
||||
* Create a Subflow object.
|
||||
* This takes a subflow definition and instance node, creates a clone of the
|
||||
* definition with unique ids applied and passes to the super class.
|
||||
* @param {[type]} parent [description]
|
||||
* @param {[type]} globalFlow [description]
|
||||
* @param {[type]} subflowDef [description]
|
||||
* @param {[type]} subflowInstance [description]
|
||||
*/
|
||||
constructor(parent,globalFlow,subflowDef,subflowInstance) {
|
||||
// console.log("CREATE SUBFLOW",subflowDef.id,subflowInstance.id,"alias?",subflowInstance._alias);
|
||||
// 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: {},
|
||||
groups: {},
|
||||
subflows: {}
|
||||
}
|
||||
|
||||
if (subflowDef.groups) {
|
||||
// Clone all of the subflow group definitions and give them new IDs
|
||||
for (i in subflowDef.groups) {
|
||||
if (subflowDef.groups.hasOwnProperty(i)) {
|
||||
node = createNodeInSubflow(subflowInstance.id,subflowDef.groups[i]);
|
||||
node_map[node._alias] = node;
|
||||
subflowInternalFlowConfig.groups[node.id] = node;
|
||||
}
|
||||
}
|
||||
}
|
||||
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 = jsonClone(subflowDef.subflows || {});
|
||||
|
||||
remapSubflowNodes(subflowInternalFlowConfig.configs,node_map);
|
||||
remapSubflowNodes(subflowInternalFlowConfig.nodes,node_map);
|
||||
remapSubflowNodes(subflowInternalFlowConfig.groups,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;
|
||||
this.path = parent.path+"/"+(subflowInstance._alias||subflowInstance.id);
|
||||
|
||||
let id = subflowInstance.id;
|
||||
if (subflowInstance._alias) {
|
||||
id = subflowInstance._alias;
|
||||
}
|
||||
this.templateCredentials = credentials.get(subflowDef.id) || {};
|
||||
this.instanceCredentials = credentials.get(id) || {};
|
||||
|
||||
var env = {};
|
||||
if (this.subflowDef.env) {
|
||||
this.subflowDef.env.forEach(e => {
|
||||
env[e.name] = e;
|
||||
if (e.type === "cred") {
|
||||
e.value = this.templateCredentials[e.name];
|
||||
}
|
||||
});
|
||||
}
|
||||
if (this.subflowInstance.env) {
|
||||
this.subflowInstance.env.forEach(e => {
|
||||
var old = env[e.name];
|
||||
var ui = old ? old.ui : null;
|
||||
env[e.name] = e;
|
||||
if (ui) {
|
||||
env[e.name].ui = ui;
|
||||
}
|
||||
if (e.type === "cred") {
|
||||
if (!old || this.instanceCredentials.hasOwnProperty(e.name) ) {
|
||||
e.value = this.instanceCredentials[e.name];
|
||||
} else if (old) {
|
||||
e.value = this.templateCredentials[e.name];
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
this.env = Object.values(env);
|
||||
}
|
||||
|
||||
/**
|
||||
* Start the subflow.
|
||||
* This creates a subflow instance node to handle the inbound messages. It also
|
||||
* rewires an subflow internal node that is connected to an output so it is connected
|
||||
* to the parent flow nodes the subflow instance is wired to.
|
||||
* @param {[type]} diff [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
async start(diff) {
|
||||
var self = this;
|
||||
// Create a subflow node to accept inbound messages and route appropriately
|
||||
var Node = require("../nodes/Node");
|
||||
|
||||
if (this.subflowDef.status) {
|
||||
var subflowStatusConfig = {
|
||||
id: this.subflowInstance.id+":status",
|
||||
type: "subflow-status",
|
||||
z: this.subflowInstance.id,
|
||||
_flow: this.parent
|
||||
}
|
||||
this.statusNode = new Node(subflowStatusConfig);
|
||||
this.statusNode.on("input", function(msg) {
|
||||
if (msg.payload !== undefined) {
|
||||
if (typeof msg.payload === "string") {
|
||||
// if msg.payload is a String, use it as status text
|
||||
self.node.status({text:msg.payload})
|
||||
return;
|
||||
} else if (Object.prototype.toString.call(msg.payload) === "[object Object]") {
|
||||
if (msg.payload.hasOwnProperty('text') || msg.payload.hasOwnProperty('fill') || msg.payload.hasOwnProperty('shape') || Object.keys(msg.payload).length === 0) {
|
||||
// msg.payload is an object that looks like a status object
|
||||
self.node.status(msg.payload);
|
||||
return;
|
||||
}
|
||||
}
|
||||
// Anything else - inspect it and use as status text
|
||||
var text = util.inspect(msg.payload);
|
||||
if (text.length > 32) { text = text.substr(0,32) + "..."; }
|
||||
self.node.status({text:text});
|
||||
} else if (msg.status !== undefined) {
|
||||
// if msg.status exists
|
||||
if (msg.status.hasOwnProperty("text") && msg.status.text.indexOf("common.") === 0) {
|
||||
msg.status.text = "node-red:"+msg.status.text;
|
||||
}
|
||||
self.node.status(msg.status)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
var subflowInstanceConfig = {
|
||||
id: this.subflowInstance.id,
|
||||
type: this.subflowInstance.type,
|
||||
g: this.subflowInstance.g,
|
||||
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 = jsonClone(subflowInstanceConfig.wires);
|
||||
}
|
||||
|
||||
this.node = new Node(subflowInstanceConfig);
|
||||
this.node.on("input", function(msg) { this.send(msg);});
|
||||
// Called when the subflow instance node is being stopped
|
||||
this.node.on("close", function(done) {
|
||||
this.status({});
|
||||
// Stop the complete subflow
|
||||
self.stop().finally(done)
|
||||
})
|
||||
this.node.status = status => this.parent.handleStatus(this.node,status);
|
||||
// Create a context instance
|
||||
// console.log("Node.context",this.type,"id:",this._alias||this.id,"z:",this.z)
|
||||
this._context = context.get(this._alias||this.id,this.z);
|
||||
|
||||
|
||||
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 = jsonClone(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 && node._originalWires) {
|
||||
node.wires = jsonClone(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];
|
||||
if (node) {
|
||||
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) {
|
||||
for (var i=0;i<this.subflowDef.out.length;i++) {
|
||||
// i: the output index
|
||||
// This is what this Output is wired to
|
||||
var 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];
|
||||
if (node) {
|
||||
if (!node._originalWires) {
|
||||
node._originalWires = jsonClone(node.wires);
|
||||
}
|
||||
node.wires[wires[j].port] = (node.wires[wires[j].port]||[]).concat(this.subflowInstance.wires[i]);
|
||||
} else {
|
||||
this.error("Unknown node referenced inside subflow: " + wires[j].id)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (this.subflowDef.status) {
|
||||
var subflowStatusId = this.statusNode.id;
|
||||
wires = this.subflowDef.status.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].push(subflowStatusId);
|
||||
this.node._updateWires(subflowInstanceConfig.wires);
|
||||
} else {
|
||||
var node = self.node_map[wires[j].id];
|
||||
if (node) {
|
||||
if (!node._originalWires) {
|
||||
node._originalWires = jsonClone(node.wires);
|
||||
}
|
||||
node.wires[wires[j].port] = (node.wires[wires[j].port]||[]);
|
||||
node.wires[wires[j].port].push(subflowStatusId);
|
||||
} else {
|
||||
this.error("Unknown node referenced inside subflow: " + wires[j].id)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return super.start(diff);
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop this subflow.
|
||||
* The `stopList` argument helps define what needs to be stopped in the case
|
||||
* of a modified-nodes/flows type deploy.
|
||||
* @param {[type]} stopList [description]
|
||||
* @param {[type]} removedList [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
stop(stopList, removedList) {
|
||||
const nodes = Object.keys(this.activeNodes);
|
||||
return super.stop(stopList, removedList).then(res => {
|
||||
nodes.forEach(id => {
|
||||
events.emit("node-status",{
|
||||
id: id
|
||||
});
|
||||
})
|
||||
return res;
|
||||
})
|
||||
|
||||
}
|
||||
/**
|
||||
* Get environment variable of subflow
|
||||
* @param {String} key name of env var
|
||||
* @return {Object} val value of env var
|
||||
*/
|
||||
getSetting(key) {
|
||||
const node = this.subflowInstance;
|
||||
if (node) {
|
||||
if (key === "NR_NODE_NAME" || key === "NR_SUBFLOW_NAME") {
|
||||
return node.name;
|
||||
}
|
||||
if (key === "NR_NODE_ID" || key === "NR_SUBFLOW_ID") {
|
||||
return node.id;
|
||||
}
|
||||
if (key === "NR_NODE_PATH" || key === "NR_SUBFLOW_PATH") {
|
||||
return node._path;
|
||||
}
|
||||
}
|
||||
if (!key.startsWith("$parent.")) {
|
||||
if (this._env.hasOwnProperty(key)) {
|
||||
return (this._env[key] && Object.hasOwn(this._env[key], 'value') && this._env[key].__clone__) ? clone(this._env[key].value) : this._env[key]
|
||||
}
|
||||
} else {
|
||||
key = key.substring(8);
|
||||
}
|
||||
// Push the request up to the parent.
|
||||
// Unlike a Flow, the parent of a Subflow could be a Group
|
||||
if (node.g) {
|
||||
return this.parent.getGroupNode(node.g).getSetting(key)
|
||||
}
|
||||
return this.parent.getSetting(key)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get a node instance from this subflow.
|
||||
* If the subflow has a status node, check for that, otherwise use
|
||||
* the super-class function
|
||||
* @param {String} id [description]
|
||||
* @param {Boolean} cancelBubble if true, prevents the flow from passing the request to the parent
|
||||
* This stops infinite loops when the parent asked this Flow for the
|
||||
* node to begin with.
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
getNode(id, cancelBubble) {
|
||||
if (this.statusNode && this.statusNode.id === id) {
|
||||
return this.statusNode;
|
||||
}
|
||||
return super.getNode(id,cancelBubble);
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle a status event from a node within this flow.
|
||||
* @param {Node} node The original node that triggered the event
|
||||
* @param {Object} statusMessage The status object
|
||||
* @param {Node} reportingNode The node emitting the status event.
|
||||
* This could be a subflow instance node when the status
|
||||
* is being delegated up.
|
||||
* @param {boolean} muteStatus Whether to emit the status event
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
handleStatus(node,statusMessage,reportingNode,muteStatus) {
|
||||
let handled = super.handleStatus(node,statusMessage,reportingNode,muteStatus);
|
||||
if (!handled) {
|
||||
if (!this.statusNode || node === this.node) {
|
||||
// No status node on this subflow caught the status message.
|
||||
// AND there is no Subflow Status node - so the user isn't
|
||||
// wanting to manage status messages themselves
|
||||
// Pass up to the parent with this subflow's instance as the
|
||||
// reporting node
|
||||
handled = this.parent.handleStatus(node,statusMessage,this.node,true);
|
||||
}
|
||||
}
|
||||
return handled;
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle an error event from a node within this flow. If there are no Catch
|
||||
* nodes within this flow, pass the event to the parent flow.
|
||||
* @param {[type]} node [description]
|
||||
* @param {[type]} logMessage [description]
|
||||
* @param {[type]} msg [description]
|
||||
* @param {[type]} reportingNode [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
handleError(node,logMessage,msg,reportingNode) {
|
||||
let handled = super.handleError(node,logMessage,msg,reportingNode);
|
||||
if (!handled) {
|
||||
// No catch node on this subflow caught the error message.
|
||||
// Pass up to the parent with the subflow's instance as the
|
||||
// reporting node.
|
||||
handled = this.parent.handleError(node,logMessage,msg,this.node);
|
||||
}
|
||||
return handled;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Clone a node definition for use within a subflow instance.
|
||||
* Give the node a new id and set its _alias property to record
|
||||
* its association with the original node definition.
|
||||
* @param {[type]} subflowInstanceId [description]
|
||||
* @param {[type]} def [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
function createNodeInSubflow(subflowInstanceId, def) {
|
||||
let node = jsonClone(def);
|
||||
let nid = `${subflowInstanceId}-${node.id}` //redUtil.generateId();
|
||||
// console.log("Create Node In subflow",node._alias, "--->",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 Complete, 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 (var j=0;j<outputs.length;j++) {
|
||||
var wires = outputs[j];
|
||||
for (var k=0;k<wires.length;k++) {
|
||||
if (nodeMap[outputs[j][k]]) {
|
||||
outputs[j][k] = nodeMap[outputs[j][k]].id
|
||||
} else {
|
||||
outputs[j][k] = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if ((node.type === 'complete' || node.type === 'catch' || node.type === 'status') && Array.isArray(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]]) {
|
||||
node[prop] = nodeMap[node[prop]].id;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class SubflowModule extends Subflow {
|
||||
/**
|
||||
* Create a Subflow Module object.
|
||||
* This is a node that has been published as a subflow.
|
||||
* @param {[type]} parent [description]
|
||||
* @param {[type]} globalFlow [description]
|
||||
* @param {[type]} subflowDef [description]
|
||||
* @param {[type]} subflowInstance [description]
|
||||
*/
|
||||
constructor(type, parent,globalFlow,subflowDef,subflowInstance) {
|
||||
super(parent,globalFlow,subflowDef,subflowInstance);
|
||||
this.TYPE = `module:${type}`;
|
||||
this.subflowType = type;
|
||||
}
|
||||
|
||||
/**
|
||||
* [log description]
|
||||
* @param {[type]} msg [description]
|
||||
* @return {[type]} [description]
|
||||
*/
|
||||
log(msg) {
|
||||
if (msg.id) {
|
||||
msg.id = this.id
|
||||
}
|
||||
if (msg.type) {
|
||||
msg.type = this.subflowType
|
||||
}
|
||||
super.log(msg);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
function createSubflow(parent,globalFlow,subflowDef,subflowInstance) {
|
||||
return new Subflow(parent,globalFlow,subflowDef,subflowInstance)
|
||||
}
|
||||
|
||||
|
||||
function createModuleInstance(type, parent,globalFlow,subflowDef,subflowInstance) {
|
||||
return new SubflowModule(type, parent,globalFlow,subflowDef,subflowInstance);
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
init: function(runtime) {},
|
||||
create: createSubflow,
|
||||
createModuleInstance: createModuleInstance
|
||||
}
|
||||
@@ -0,0 +1,876 @@
|
||||
/**
|
||||
* 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 jsonClone = require("rfdc")();
|
||||
|
||||
var Flow = require('./Flow');
|
||||
|
||||
var typeRegistry = require("@node-red/registry");
|
||||
var deprecated = typeRegistry.deprecated;
|
||||
|
||||
var context = require("../nodes/context")
|
||||
var credentials = require("../nodes/credentials");
|
||||
var flowUtil = require("./util");
|
||||
var log;
|
||||
const events = require("@node-red/util").events;
|
||||
var redUtil = require("@node-red/util").util;
|
||||
|
||||
var storage = null;
|
||||
var settings = null;
|
||||
|
||||
var activeConfig = null;
|
||||
var activeFlowConfig = null;
|
||||
|
||||
var activeFlows = {};
|
||||
var started = false;
|
||||
var state = 'stop'
|
||||
|
||||
var credentialsPendingReset = false;
|
||||
|
||||
var activeNodesToFlow = {};
|
||||
|
||||
var typeEventRegistered = false;
|
||||
|
||||
function init(runtime) {
|
||||
if (started) {
|
||||
throw new Error("Cannot init without a stop");
|
||||
}
|
||||
settings = runtime.settings;
|
||||
storage = runtime.storage;
|
||||
log = runtime.log;
|
||||
started = false;
|
||||
state = 'stop';
|
||||
if (!typeEventRegistered) {
|
||||
events.on('type-registered',function(type) {
|
||||
if (activeFlowConfig && activeFlowConfig.missingTypes.length > 0) {
|
||||
var i = activeFlowConfig.missingTypes.indexOf(type);
|
||||
if (i != -1) {
|
||||
log.info(log._("nodes.flows.registered-missing", {type:type}));
|
||||
activeFlowConfig.missingTypes.splice(i,1);
|
||||
if (activeFlowConfig.missingTypes.length === 0 && started) {
|
||||
events.emit("runtime-event",{id:"runtime-state",retain: true});
|
||||
start();
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
typeEventRegistered = true;
|
||||
}
|
||||
Flow.init(runtime);
|
||||
flowUtil.init(runtime);
|
||||
}
|
||||
|
||||
function loadFlows() {
|
||||
var config;
|
||||
return storage.getFlows().then(function(_config) {
|
||||
config = _config;
|
||||
log.debug("loaded flow revision: "+config.rev);
|
||||
return credentials.load(config.credentials).then(function() {
|
||||
events.emit("runtime-event",{id:"runtime-state",retain:true});
|
||||
return config;
|
||||
});
|
||||
}).catch(function(err) {
|
||||
if (err.code === "credentials_load_failed" && !storage.projects) {
|
||||
// project disabled, credential load failed
|
||||
credentialsPendingReset = true;
|
||||
log.warn(log._("nodes.flows.error",{message:err.toString()}));
|
||||
events.emit("runtime-event",{id:"runtime-state",payload:{type:"warning",error:err.code,text:"notification.warnings.credentials_load_failed_reset"},retain:true});
|
||||
return config;
|
||||
} else {
|
||||
activeConfig = null;
|
||||
events.emit("runtime-event",{id:"runtime-state",payload:{type:"warning",error:err.code,project:err.project,text:"notification.warnings."+err.code},retain:true});
|
||||
if (err.code === "project_not_found") {
|
||||
log.warn(log._("storage.localfilesystem.projects.project-not-found",{project:err.project}));
|
||||
} else {
|
||||
log.warn(log._("nodes.flows.error",{message:err.toString()}));
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
});
|
||||
}
|
||||
function load(forceStart) {
|
||||
if (forceStart && settings.safeMode) {
|
||||
// This is a force reload from the API - disable safeMode
|
||||
delete settings.safeMode;
|
||||
}
|
||||
return setFlows(null,null,"load",false,forceStart);
|
||||
}
|
||||
|
||||
/*
|
||||
* _config - new node array configuration
|
||||
* _credentials - new credentials configuration (optional)
|
||||
* type - full/nodes/flows/load (default full)
|
||||
* muteLog - don't emit the standard log messages (used for individual flow api)
|
||||
*/
|
||||
function setFlows(_config,_credentials,type,muteLog,forceStart,user) {
|
||||
if (typeof _credentials === "string") {
|
||||
type = _credentials;
|
||||
_credentials = null;
|
||||
}
|
||||
type = type||"full";
|
||||
if (settings.safeMode) {
|
||||
if (type !== "load") {
|
||||
// If in safeMode, the flows are stopped. We cannot do a modified nodes/flows
|
||||
// type deploy as nothing is running. Can only do a "load" or "full" deploy.
|
||||
// The "load" case is already handled in `load()` to distinguish between
|
||||
// startup-load and api-request-load.
|
||||
type = "full";
|
||||
delete settings.safeMode;
|
||||
}
|
||||
}
|
||||
|
||||
var configSavePromise = null;
|
||||
var config = null;
|
||||
var diff;
|
||||
var newFlowConfig;
|
||||
var isLoad = false;
|
||||
if (type === "load") {
|
||||
isLoad = true;
|
||||
configSavePromise = loadFlows().then(function(_config) {
|
||||
config = jsonClone(_config.flows);
|
||||
newFlowConfig = flowUtil.parseConfig(jsonClone(config));
|
||||
type = "full";
|
||||
return _config.rev;
|
||||
});
|
||||
} else {
|
||||
// Clone the provided config so it can be manipulated
|
||||
config = jsonClone(_config);
|
||||
// Parse the configuration
|
||||
newFlowConfig = flowUtil.parseConfig(jsonClone(config));
|
||||
// Generate a diff to identify what has changed
|
||||
diff = flowUtil.diffConfigs(activeFlowConfig,newFlowConfig);
|
||||
|
||||
// Now the flows have been compared, remove any credentials from newFlowConfig
|
||||
// so they don't cause false-positive diffs the next time a flow is deployed
|
||||
for (var id in newFlowConfig.allNodes) {
|
||||
if (newFlowConfig.allNodes.hasOwnProperty(id)) {
|
||||
delete newFlowConfig.allNodes[id].credentials;
|
||||
}
|
||||
}
|
||||
var credsDirty;
|
||||
|
||||
if (_credentials) {
|
||||
if (_credentials['$']) {
|
||||
// this is a set of encrypted credentials - pass to load to decrypt
|
||||
// the complete set
|
||||
configSavePromise = credentials.load(_credentials);
|
||||
} else {
|
||||
credentials.clean(config);
|
||||
// A full set of credentials have been provided. Use those instead
|
||||
let credentialSavePromises = [];
|
||||
for (let id in _credentials) {
|
||||
if (_credentials.hasOwnProperty(id)) {
|
||||
credentialSavePromises.push(credentials.add(id,_credentials[id]));
|
||||
}
|
||||
}
|
||||
configSavePromise = Promise.all(credentialSavePromises);
|
||||
credsDirty = true;
|
||||
}
|
||||
} else {
|
||||
// Allow the credential store to remove anything no longer needed
|
||||
credentials.clean(config);
|
||||
|
||||
// Remember whether credentials need saving or not
|
||||
var credsDirty = credentials.dirty();
|
||||
|
||||
configSavePromise = Promise.resolve();
|
||||
}
|
||||
|
||||
// Get the latest credentials and ask storage to save them (if needed)
|
||||
// as well as the new flow configuration.
|
||||
configSavePromise = configSavePromise.then(function() {
|
||||
return credentials.export()
|
||||
}).then(function(creds) {
|
||||
var saveConfig = {
|
||||
flows: config,
|
||||
credentialsDirty:credsDirty,
|
||||
credentials: creds
|
||||
}
|
||||
return storage.saveFlows(saveConfig, user);
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
return configSavePromise.then(flowRevision => {
|
||||
if (!isLoad) {
|
||||
log.debug("saved flow revision: "+flowRevision);
|
||||
}
|
||||
activeConfig = {
|
||||
flows:config,
|
||||
rev:flowRevision
|
||||
};
|
||||
activeFlowConfig = newFlowConfig;
|
||||
if (forceStart || started) {
|
||||
// Flows are running (or should be)
|
||||
|
||||
// Stop the active flows (according to deploy type and the diff)
|
||||
return stop(type,diff,muteLog,true).then(() => {
|
||||
// Once stopped, allow context to remove anything no longer needed
|
||||
return context.clean(activeFlowConfig)
|
||||
}).then(() => {
|
||||
if (!isLoad) {
|
||||
log.info(log._("nodes.flows.updated-flows"));
|
||||
}
|
||||
// Start the active flows
|
||||
start(type,diff,muteLog,true).then(() => {
|
||||
events.emit("runtime-event",{id:"runtime-deploy",payload:{revision:flowRevision},retain: true});
|
||||
});
|
||||
// Return the new revision asynchronously to the actual start
|
||||
return flowRevision;
|
||||
}).catch(function(err) { })
|
||||
} else {
|
||||
if (!isLoad) {
|
||||
log.info(log._("nodes.flows.updated-flows"));
|
||||
}
|
||||
events.emit("runtime-event",{id:"runtime-deploy",payload:{revision:flowRevision},retain: true});
|
||||
return flowRevision;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function getNode(id) {
|
||||
var node;
|
||||
if (activeNodesToFlow[id] && activeFlows[activeNodesToFlow[id]]) {
|
||||
return activeFlows[activeNodesToFlow[id]].getNode(id,true);
|
||||
}
|
||||
for (var flowId in activeFlows) {
|
||||
if (activeFlows.hasOwnProperty(flowId)) {
|
||||
node = activeFlows[flowId].getNode(id,true);
|
||||
if (node) {
|
||||
return node;
|
||||
}
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function eachNode(cb) {
|
||||
for (var id in activeFlowConfig.allNodes) {
|
||||
if (activeFlowConfig.allNodes.hasOwnProperty(id)) {
|
||||
cb(activeFlowConfig.allNodes[id]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function getFlows() {
|
||||
return activeConfig;
|
||||
}
|
||||
|
||||
async function start(type,diff,muteLog,isDeploy) {
|
||||
type = type || "full";
|
||||
if (diff && diff.globalConfigChanged) {
|
||||
type = 'full'
|
||||
}
|
||||
|
||||
started = true;
|
||||
state = 'start'
|
||||
var i;
|
||||
// If there are missing types, report them, emit the necessary runtime event and return
|
||||
if (activeFlowConfig.missingTypes.length > 0) {
|
||||
log.info(log._("nodes.flows.missing-types"));
|
||||
var knownUnknowns = 0;
|
||||
for (i=0;i<activeFlowConfig.missingTypes.length;i++) {
|
||||
var nodeType = activeFlowConfig.missingTypes[i];
|
||||
var info = deprecated.get(nodeType);
|
||||
if (info) {
|
||||
log.info(log._("nodes.flows.missing-type-provided",{type:activeFlowConfig.missingTypes[i],module:info.module}));
|
||||
knownUnknowns += 1;
|
||||
} else {
|
||||
log.info(" - "+activeFlowConfig.missingTypes[i]);
|
||||
}
|
||||
}
|
||||
if (knownUnknowns > 0) {
|
||||
log.info(log._("nodes.flows.missing-type-install-1"));
|
||||
log.info(" npm install <module name>");
|
||||
log.info(log._("nodes.flows.missing-type-install-2"));
|
||||
log.info(" "+settings.userDir);
|
||||
}
|
||||
events.emit("runtime-event",{id:"runtime-state",payload:{state: 'stop', error:"missing-types", type:"warning",text:"notification.warnings.missing-types",types:activeFlowConfig.missingTypes},retain:true});
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await typeRegistry.checkFlowDependencies(activeConfig.flows);
|
||||
} catch(err) {
|
||||
log.info("Failed to load external modules required by this flow:");
|
||||
const missingModules = [];
|
||||
for (i=0;i<err.length;i++) {
|
||||
let errMessage = err[i].error.toString()
|
||||
missingModules.push({module:err[i].module.module, error: err[i].error.code || err[i].error.toString()})
|
||||
log.info(` - ${err[i].module.spec} [${err[i].error.code || "unknown_error"}]`);
|
||||
}
|
||||
events.emit("runtime-event",{id:"runtime-state",payload:{state: 'stop', error:"missing-modules", type:"warning",text:"notification.warnings.missing-modules",modules:missingModules},retain:true});
|
||||
return;
|
||||
}
|
||||
|
||||
// In safe mode, don't actually start anything, emit the necessary runtime event and return
|
||||
if (settings.safeMode) {
|
||||
log.info("*****************************************************************")
|
||||
log.info(log._("nodes.flows.safe-mode"));
|
||||
log.info("*****************************************************************")
|
||||
state = 'safe'
|
||||
events.emit("runtime-event",{id:"runtime-state",payload:{state: 'safe', error:"safe-mode", type:"warning",text:"notification.warnings.safe-mode"},retain:true});
|
||||
return;
|
||||
}
|
||||
|
||||
let runtimeState
|
||||
try {
|
||||
runtimeState = settings.get('runtimeFlowState') || 'start'
|
||||
} catch (err) {}
|
||||
if (runtimeState === 'stop') {
|
||||
log.info(log._("nodes.flows.stopped-flows"));
|
||||
events.emit("runtime-event",{id:"runtime-state",payload:{ state: 'stop', deploy:isDeploy },retain:true});
|
||||
state = 'stop'
|
||||
started = false
|
||||
return
|
||||
}
|
||||
|
||||
if (!muteLog) {
|
||||
if (type !== "full") {
|
||||
log.info(log._("nodes.flows.starting-modified-"+type));
|
||||
} else {
|
||||
log.info(log._("nodes.flows.starting-flows"));
|
||||
}
|
||||
}
|
||||
|
||||
events.emit("flows:starting", {config: activeConfig, type: type, diff: diff})
|
||||
|
||||
var id;
|
||||
if (type === "full") {
|
||||
// A full start means everything should
|
||||
|
||||
// Check the 'global' flow is running
|
||||
if (!activeFlows['global']) {
|
||||
log.debug("red/nodes/flows.start : starting flow : global");
|
||||
activeFlows['global'] = Flow.create(flowAPI,activeFlowConfig);
|
||||
}
|
||||
|
||||
// Check each flow in the active configuration
|
||||
for (id in activeFlowConfig.flows) {
|
||||
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(activeFlows['global'],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);
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// A modified-type deploy means restarting things that have changed
|
||||
|
||||
// Update the global flow
|
||||
if (activeFlows['global']) {
|
||||
activeFlows['global'].update(activeFlowConfig,activeFlowConfig);
|
||||
} else {
|
||||
log.debug("red/nodes/flows.start : starting flow : global");
|
||||
activeFlows['global'] = Flow.create(flowAPI,activeFlowConfig);
|
||||
}
|
||||
for (id in activeFlowConfig.flows) {
|
||||
if (activeFlowConfig.flows.hasOwnProperty(id)) {
|
||||
if (!activeFlowConfig.flows[id].disabled) {
|
||||
if (activeFlows[id]) {
|
||||
// This flow exists and is not disabled, so update it
|
||||
activeFlows[id].update(activeFlowConfig,activeFlowConfig.flows[id]);
|
||||
} else {
|
||||
// This flow didn't previously exist, so create it
|
||||
activeFlows[id] = Flow.create(activeFlows['global'],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);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
for (id in activeFlows) {
|
||||
if (activeFlows.hasOwnProperty(id)) {
|
||||
try {
|
||||
await 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("flows:started", {config: activeConfig, type: type, diff: diff});
|
||||
// Deprecated event
|
||||
events.emit("nodes-started");
|
||||
|
||||
if (credentialsPendingReset === true) {
|
||||
credentialsPendingReset = false;
|
||||
} else {
|
||||
events.emit("runtime-event",{id:"runtime-state", payload:{ state: 'start', deploy:isDeploy}, retain:true});
|
||||
}
|
||||
|
||||
if (!muteLog) {
|
||||
if (type !== "full") {
|
||||
log.info(log._("nodes.flows.started-modified-"+type));
|
||||
} else {
|
||||
log.info(log._("nodes.flows.started-flows"));
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
function stop(type,diff,muteLog,isDeploy) {
|
||||
if (!started) {
|
||||
return Promise.resolve();
|
||||
}
|
||||
type = type||"full";
|
||||
diff = diff||{
|
||||
added:[],
|
||||
changed:[],
|
||||
removed:[],
|
||||
rewired:[],
|
||||
linked:[],
|
||||
flowChanged:[]
|
||||
};
|
||||
if (!muteLog) {
|
||||
if (type !== "full") {
|
||||
log.info(log._("nodes.flows.stopping-modified-"+type));
|
||||
} else {
|
||||
log.info(log._("nodes.flows.stopping-flows"));
|
||||
}
|
||||
}
|
||||
if (diff.globalConfigChanged) {
|
||||
type = 'full'
|
||||
}
|
||||
started = false;
|
||||
state = 'stop'
|
||||
var promises = [];
|
||||
var stopList;
|
||||
var removedList = diff.removed;
|
||||
if (type === 'nodes') {
|
||||
stopList = diff.changed.concat(diff.removed);
|
||||
} else if (type === 'flows') {
|
||||
stopList = diff.changed.concat(diff.removed).concat(diff.linked).concat(diff.rewired);
|
||||
}
|
||||
events.emit("flows:stopping",{config: activeConfig, type: type, diff: diff})
|
||||
|
||||
// Stop the global flow object last
|
||||
let activeFlowIds = Object.keys(activeFlows);
|
||||
let globalIndex = activeFlowIds.indexOf("global");
|
||||
if (globalIndex !== -1) {
|
||||
activeFlowIds.splice(globalIndex,1);
|
||||
activeFlowIds.push("global");
|
||||
}
|
||||
|
||||
activeFlowIds.forEach(id => {
|
||||
if (activeFlows.hasOwnProperty(id)) {
|
||||
var flowStateChanged = diff && (diff.flowChanged.indexOf(id) !== -1 || diff.added.indexOf(id) !== -1 || diff.removed.indexOf(id) !== -1);
|
||||
log.debug("red/nodes/flows.stop : stopping flow : "+id);
|
||||
promises.push(activeFlows[id].stop(flowStateChanged?null:stopList,removedList));
|
||||
if (type === "full" || flowStateChanged || diff.removed.indexOf(id)!==-1) {
|
||||
delete activeFlows[id];
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
return Promise.all(promises).then(function() {
|
||||
for (let id in activeNodesToFlow) {
|
||||
if (activeNodesToFlow.hasOwnProperty(id)) {
|
||||
if (!activeFlows[activeNodesToFlow[id]]) {
|
||||
delete activeNodesToFlow[id];
|
||||
}
|
||||
}
|
||||
}
|
||||
if (stopList) {
|
||||
stopList.forEach(function(id) {
|
||||
delete activeNodesToFlow[id];
|
||||
});
|
||||
}
|
||||
if (!muteLog) {
|
||||
if (type !== "full") {
|
||||
log.info(log._("nodes.flows.stopped-modified-"+type));
|
||||
} else {
|
||||
log.info(log._("nodes.flows.stopped-flows"));
|
||||
}
|
||||
}
|
||||
events.emit("flows:stopped",{config: activeConfig, type: type, diff: diff});
|
||||
|
||||
events.emit("runtime-event",{ id:"runtime-state", payload:{ state: 'stop', deploy:isDeploy }, retain:true });
|
||||
// Deprecated event
|
||||
events.emit("nodes-stopped");
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
function checkTypeInUse(id) {
|
||||
var nodeInfo = typeRegistry.getNodeInfo(id);
|
||||
if (!nodeInfo) {
|
||||
throw new Error(log._("nodes.index.unrecognised-id", {id:id}));
|
||||
} else {
|
||||
var inUse = {};
|
||||
var config = getFlows();
|
||||
config.flows.forEach(function(n) {
|
||||
inUse[n.type] = (inUse[n.type]||0)+1;
|
||||
});
|
||||
var nodesInUse = [];
|
||||
nodeInfo.types.forEach(function(t) {
|
||||
if (inUse[t]) {
|
||||
nodesInUse.push(t);
|
||||
}
|
||||
});
|
||||
if (nodesInUse.length > 0) {
|
||||
var msg = nodesInUse.join(", ");
|
||||
var err = new Error(log._("nodes.index.type-in-use", {msg:msg}));
|
||||
err.code = "type_in_use";
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function updateMissingTypes() {
|
||||
var subflowInstanceRE = /^subflow:(.+)$/;
|
||||
activeFlowConfig.missingTypes = [];
|
||||
|
||||
for (var id in activeFlowConfig.allNodes) {
|
||||
if (activeFlowConfig.allNodes.hasOwnProperty(id)) {
|
||||
var node = activeFlowConfig.allNodes[id];
|
||||
if (node.type !== 'tab' && node.type !== 'subflow') {
|
||||
var subflowDetails = subflowInstanceRE.exec(node.type);
|
||||
if ( (subflowDetails && !activeFlowConfig.subflows[subflowDetails[1]]) || (!subflowDetails && !typeRegistry.get(node.type)) ) {
|
||||
if (activeFlowConfig.missingTypes.indexOf(node.type) === -1) {
|
||||
activeFlowConfig.missingTypes.push(node.type);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function addFlow(flow, user) {
|
||||
var i,node;
|
||||
if (!flow.hasOwnProperty('nodes')) {
|
||||
throw new Error('missing nodes property');
|
||||
}
|
||||
flow.id = redUtil.generateId();
|
||||
|
||||
var tabNode = {
|
||||
type:'tab',
|
||||
label:flow.label,
|
||||
id:flow.id
|
||||
}
|
||||
if (flow.hasOwnProperty('info')) {
|
||||
tabNode.info = flow.info;
|
||||
}
|
||||
if (flow.hasOwnProperty('disabled')) {
|
||||
tabNode.disabled = flow.disabled;
|
||||
}
|
||||
if (flow.hasOwnProperty('env')) {
|
||||
tabNode.env = flow.env;
|
||||
}
|
||||
|
||||
var nodes = [tabNode];
|
||||
|
||||
for (i=0;i<flow.nodes.length;i++) {
|
||||
node = flow.nodes[i];
|
||||
if (activeFlowConfig.allNodes[node.id]) {
|
||||
// TODO nls
|
||||
throw new Error('duplicate id');
|
||||
}
|
||||
if (node.type === 'tab' || node.type === 'subflow') {
|
||||
throw new Error('invalid node type: '+node.type);
|
||||
}
|
||||
node.z = flow.id;
|
||||
nodes.push(node);
|
||||
}
|
||||
if (flow.configs) {
|
||||
for (i=0;i<flow.configs.length;i++) {
|
||||
node = flow.configs[i];
|
||||
if (activeFlowConfig.allNodes[node.id]) {
|
||||
// TODO nls
|
||||
throw new Error('duplicate id');
|
||||
}
|
||||
if (node.type === 'tab' || node.type === 'subflow') {
|
||||
throw new Error('invalid node type: '+node.type);
|
||||
}
|
||||
node.z = flow.id;
|
||||
nodes.push(node);
|
||||
}
|
||||
}
|
||||
var newConfig = jsonClone(activeConfig.flows);
|
||||
newConfig = newConfig.concat(nodes);
|
||||
|
||||
return setFlows(newConfig, null, 'flows', true, null, user).then(function() {
|
||||
log.info(log._("nodes.flows.added-flow",{label:(flow.label?flow.label+" ":"")+"["+flow.id+"]"}));
|
||||
return flow.id;
|
||||
});
|
||||
}
|
||||
|
||||
function getFlow(id) {
|
||||
var flow;
|
||||
if (id === 'global') {
|
||||
flow = activeFlowConfig;
|
||||
} else {
|
||||
flow = activeFlowConfig.flows[id];
|
||||
}
|
||||
if (!flow) {
|
||||
return null;
|
||||
}
|
||||
var result = {
|
||||
id: id
|
||||
};
|
||||
if (flow.label) {
|
||||
result.label = flow.label;
|
||||
}
|
||||
if (flow.hasOwnProperty('disabled')) {
|
||||
result.disabled = flow.disabled;
|
||||
}
|
||||
if (flow.hasOwnProperty('info')) {
|
||||
result.info = flow.info;
|
||||
}
|
||||
if (flow.hasOwnProperty('env')) {
|
||||
result.env = flow.env;
|
||||
}
|
||||
if (id !== 'global') {
|
||||
result.nodes = [];
|
||||
}
|
||||
|
||||
if (flow.groups) {
|
||||
var nodeIds = Object.keys(flow.groups);
|
||||
if (nodeIds.length > 0) {
|
||||
nodeIds.forEach(function(nodeId) {
|
||||
var node = jsonClone(flow.groups[nodeId]);
|
||||
delete node.credentials;
|
||||
result.nodes.push(node)
|
||||
})
|
||||
}
|
||||
}
|
||||
if (flow.nodes) {
|
||||
var nodeIds = Object.keys(flow.nodes);
|
||||
if (nodeIds.length > 0) {
|
||||
nodeIds.forEach(function(nodeId) {
|
||||
var node = jsonClone(flow.nodes[nodeId]);
|
||||
if (node.type === 'link out') {
|
||||
delete node.wires;
|
||||
}
|
||||
delete node.credentials;
|
||||
result.nodes.push(node)
|
||||
})
|
||||
}
|
||||
}
|
||||
if (flow.configs) {
|
||||
var configIds = Object.keys(flow.configs);
|
||||
result.configs = configIds.map(function(configId) {
|
||||
const node = jsonClone(flow.configs[configId]);
|
||||
delete node.credentials;
|
||||
return node
|
||||
|
||||
})
|
||||
if (result.configs.length === 0) {
|
||||
delete result.configs;
|
||||
}
|
||||
}
|
||||
if (flow.subflows) {
|
||||
var subflowIds = Object.keys(flow.subflows);
|
||||
result.subflows = subflowIds.map(function(subflowId) {
|
||||
var subflow = jsonClone(flow.subflows[subflowId]);
|
||||
var nodeIds = Object.keys(subflow.nodes);
|
||||
subflow.nodes = nodeIds.map(function(id) {
|
||||
const node = jsonClone(subflow.nodes[id])
|
||||
delete node.credentials
|
||||
return node
|
||||
});
|
||||
if (subflow.groups) {
|
||||
var nodeIds = Object.keys(subflow.groups);
|
||||
if (nodeIds.length > 0) {
|
||||
nodeIds.forEach(function(nodeId) {
|
||||
var node = jsonClone(subflow.groups[nodeId]);
|
||||
delete node.credentials;
|
||||
subflow.nodes.push(node)
|
||||
})
|
||||
}
|
||||
delete subflow.groups
|
||||
}
|
||||
if (subflow.configs) {
|
||||
var configIds = Object.keys(subflow.configs);
|
||||
subflow.configs = configIds.map(function(id) {
|
||||
const node = jsonClone(subflow.configs[id])
|
||||
delete node.credentials
|
||||
return node
|
||||
})
|
||||
}
|
||||
delete subflow.instances;
|
||||
return subflow;
|
||||
});
|
||||
if (result.subflows.length === 0) {
|
||||
delete result.subflows;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
async function updateFlow(id,newFlow, user) {
|
||||
var label = id;
|
||||
if (id !== 'global') {
|
||||
if (!activeFlowConfig.flows[id]) {
|
||||
var e = new Error();
|
||||
e.code = 404;
|
||||
throw e;
|
||||
}
|
||||
label = activeFlowConfig.flows[id].label;
|
||||
}
|
||||
var newConfig = jsonClone(activeConfig.flows);
|
||||
var nodes;
|
||||
|
||||
if (id === 'global') {
|
||||
// Remove all nodes whose z is not a known flow
|
||||
// When subflows can be owned by a flow, this logic will have to take
|
||||
// that into account
|
||||
newConfig = newConfig.filter(function(node) {
|
||||
return node.type === 'tab' || (node.hasOwnProperty('z') && activeFlowConfig.flows.hasOwnProperty(node.z));
|
||||
})
|
||||
|
||||
// Add in the new config nodes
|
||||
nodes = newFlow.configs||[];
|
||||
if (newFlow.subflows) {
|
||||
// Add in the new subflows
|
||||
newFlow.subflows.forEach(function(sf) {
|
||||
nodes = nodes.concat(sf.nodes||[]).concat(sf.configs||[]);
|
||||
delete sf.nodes;
|
||||
delete sf.configs;
|
||||
nodes.push(sf);
|
||||
});
|
||||
}
|
||||
} else {
|
||||
newConfig = newConfig.filter(function(node) {
|
||||
return node.z !== id && node.id !== id;
|
||||
});
|
||||
var tabNode = {
|
||||
type:'tab',
|
||||
label:newFlow.label,
|
||||
id:id
|
||||
}
|
||||
if (newFlow.hasOwnProperty('info')) {
|
||||
tabNode.info = newFlow.info;
|
||||
}
|
||||
if (newFlow.hasOwnProperty('disabled')) {
|
||||
tabNode.disabled = newFlow.disabled;
|
||||
}
|
||||
if (newFlow.hasOwnProperty('env')) {
|
||||
tabNode.env = newFlow.env;
|
||||
}
|
||||
if (newFlow.hasOwnProperty('credentials')) {
|
||||
tabNode.credentials = newFlow.credentials;
|
||||
}
|
||||
|
||||
nodes = [tabNode].concat(newFlow.nodes||[]).concat(newFlow.configs||[]);
|
||||
nodes.forEach(function(n) {
|
||||
if (n.type !== 'tab') {
|
||||
n.z = id;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
newConfig = newConfig.concat(nodes);
|
||||
return setFlows(newConfig, null, 'flows', true, null, user).then(function() {
|
||||
log.info(log._("nodes.flows.updated-flow",{label:(label?label+" ":"")+"["+id+"]"}));
|
||||
})
|
||||
}
|
||||
|
||||
async function removeFlow(id, user) {
|
||||
if (id === 'global') {
|
||||
// TODO: nls + error code
|
||||
throw new Error('not allowed to remove global');
|
||||
}
|
||||
var flow = activeFlowConfig.flows[id];
|
||||
if (!flow) {
|
||||
var e = new Error();
|
||||
e.code = 404;
|
||||
throw e;
|
||||
}
|
||||
|
||||
var newConfig = jsonClone(activeConfig.flows);
|
||||
newConfig = newConfig.filter(function(node) {
|
||||
return node.z !== id && node.id !== id;
|
||||
});
|
||||
|
||||
return setFlows(newConfig, null, 'flows', true, null, user).then(function() {
|
||||
log.info(log._("nodes.flows.removed-flow",{label:(flow.label?flow.label+" ":"")+"["+flow.id+"]"}));
|
||||
});
|
||||
}
|
||||
|
||||
const flowAPI = {
|
||||
getNode: getNode,
|
||||
handleError: () => false,
|
||||
handleStatus: () => false,
|
||||
getSetting: k => flowUtil.getEnvVar(k),
|
||||
log: m => log.log(m)
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
init: init,
|
||||
|
||||
/**
|
||||
* Load the current flow configuration from storage
|
||||
* @return a promise for the loading of the config
|
||||
*/
|
||||
load: load,
|
||||
loadFlows: load,
|
||||
|
||||
get:getNode,
|
||||
eachNode: eachNode,
|
||||
|
||||
/**
|
||||
* Gets the current flow configuration
|
||||
*/
|
||||
getFlows: getFlows,
|
||||
|
||||
/**
|
||||
* Sets the current active config.
|
||||
* @param config the configuration to enable
|
||||
* @param type the type of deployment to do: full (default), nodes, flows, load
|
||||
* @return a promise for the saving/starting of the new flow
|
||||
*/
|
||||
setFlows: setFlows,
|
||||
|
||||
/**
|
||||
* Starts the current flow configuration
|
||||
*/
|
||||
startFlows: start,
|
||||
|
||||
/**
|
||||
* Stops the current flow configuration
|
||||
* @return a promise for the stopping of the flow
|
||||
*/
|
||||
stopFlows: stop,
|
||||
|
||||
get started() { return started },
|
||||
state: () => { return state },
|
||||
// handleError: handleError,
|
||||
// handleStatus: handleStatus,
|
||||
|
||||
checkTypeInUse: checkTypeInUse,
|
||||
|
||||
addFlow: addFlow,
|
||||
getFlow: getFlow,
|
||||
updateFlow: updateFlow,
|
||||
removeFlow: removeFlow,
|
||||
disableFlow:null,
|
||||
enableFlow:null,
|
||||
isDeliveryModeAsync: function() {
|
||||
// If settings is null, this is likely being run by unit tests
|
||||
return !settings || !settings.runtimeSyncDelivery
|
||||
}
|
||||
|
||||
};
|
||||
@@ -0,0 +1,735 @@
|
||||
/**
|
||||
* 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 jsonClone = require("rfdc")();
|
||||
const redUtil = require("@node-red/util").util;
|
||||
const Log = require("@node-red/util").log;
|
||||
const typeRegistry = require("@node-red/registry");
|
||||
const subflowInstanceRE = /^subflow:(.+)$/;
|
||||
|
||||
let _runtime = null;
|
||||
let envVarExcludes = {};
|
||||
|
||||
function init(runtime) {
|
||||
_runtime = runtime;
|
||||
envVarExcludes = {};
|
||||
if (runtime.settings.hasOwnProperty('envVarExcludes') && Array.isArray(runtime.settings.envVarExcludes)) {
|
||||
runtime.settings.envVarExcludes.forEach(v => envVarExcludes[v] = true);
|
||||
}
|
||||
}
|
||||
|
||||
function diffNodes(oldNode,newNode) {
|
||||
if (oldNode == null) {
|
||||
return true;
|
||||
}
|
||||
const keyFilter = p => p != 'x' && p != 'y' && p != 'wires'
|
||||
const groupKeyFilter = p => keyFilter(p) && p != 'nodes' && p != 'style' && p != 'w' && p != 'h'
|
||||
var oldKeys = Object.keys(oldNode).filter(oldNode.type === 'group' ? groupKeyFilter : keyFilter);
|
||||
var newKeys = Object.keys(newNode).filter(newNode.type === 'group' ? groupKeyFilter : keyFilter);
|
||||
|
||||
if (oldKeys.length != newKeys.length) {
|
||||
return true;
|
||||
}
|
||||
for (var i=0;i<newKeys.length;i++) {
|
||||
var p = newKeys[i];
|
||||
if (!redUtil.compareObjects(oldNode[p],newNode[p])) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
var EnvVarPropertyRE_old = /^\$\((\S+)\)$/;
|
||||
var EnvVarPropertyRE = /^\${(\S+)}$/;
|
||||
|
||||
|
||||
function mapEnvVarProperties(obj,prop,flow,config) {
|
||||
const v = obj[prop];
|
||||
if (Buffer.isBuffer(v)) {
|
||||
return;
|
||||
} else if (Array.isArray(v)) {
|
||||
for (let i=0;i<v.length;i++) {
|
||||
mapEnvVarProperties(v,i,flow,config);
|
||||
}
|
||||
} else if (typeof obj[prop] === 'string') {
|
||||
if (obj[prop][0] === "$" && (EnvVarPropertyRE_old.test(v) || EnvVarPropertyRE.test(v)) ) {
|
||||
const envVar = v.substring(2,v.length-1);
|
||||
const r = redUtil.getSetting(config, envVar, flow);
|
||||
if (r !== undefined) {
|
||||
obj[prop] = r
|
||||
}
|
||||
}
|
||||
} else {
|
||||
for (var p in v) {
|
||||
if (v.hasOwnProperty(p)) {
|
||||
mapEnvVarProperties(v,p,flow,config);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function evaluateEnvProperties(flow, env, credentials) {
|
||||
credentials = credentials || {}
|
||||
const pendingEvaluations = []
|
||||
const evaluatedEnv = {}
|
||||
const envTypes = []
|
||||
for (let i = 0; i < env.length; i++) {
|
||||
let { name, value, type } = env[i]
|
||||
if (type === "env") {
|
||||
// Do env types last as they may include references to other env vars
|
||||
// at this level which need to be resolved before they can be looked-up
|
||||
envTypes.push(env[i])
|
||||
} else if (type === "bool") {
|
||||
value = (value === "true") || (value === true);
|
||||
} else if (type === "cred") {
|
||||
if (credentials.hasOwnProperty(name)) {
|
||||
value = credentials[name];
|
||||
}
|
||||
} else if (type ==='jsonata') {
|
||||
pendingEvaluations.push(new Promise((resolve, _) => {
|
||||
redUtil.evaluateNodeProperty(value, 'jsonata',{
|
||||
// Fake a node object to provide access to _flow and context
|
||||
_flow: flow,
|
||||
context: () => {
|
||||
return {
|
||||
flow: {
|
||||
get: (value, store, callback) => {
|
||||
return flow.getContext('flow').get(value, store, callback)
|
||||
}
|
||||
},
|
||||
global: {
|
||||
get: (value, store, callback) => {
|
||||
return flow.getContext('global').get(value, store, callback)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}, null, (err, result) => {
|
||||
if (!err) {
|
||||
if (typeof result === 'object') {
|
||||
result = { value: result, __clone__: true}
|
||||
}
|
||||
evaluatedEnv[name] = result
|
||||
} else {
|
||||
evaluatedEnv[name] = undefined
|
||||
flow.error(`Error evaluating env property '${name}': ${err.toString()}`)
|
||||
}
|
||||
resolve()
|
||||
});
|
||||
}))
|
||||
} else if (type === "conf-type" && /^\${[^}]+}$/.test(value)) {
|
||||
// Get the config node from the parent subflow
|
||||
const name = value.substring(2, value.length - 1);
|
||||
value = flow.getSetting(name);
|
||||
} else {
|
||||
try {
|
||||
value = redUtil.evaluateNodeProperty(value, type, {_flow: flow}, null, null);
|
||||
if (typeof value === 'object') {
|
||||
value = { value: value, __clone__: true}
|
||||
}
|
||||
} catch (err) {
|
||||
value = undefined
|
||||
flow.error(`Error evaluating env property '${name}': ${err.toString()}`)
|
||||
}
|
||||
}
|
||||
evaluatedEnv[name] = value
|
||||
}
|
||||
if (pendingEvaluations.length > 0) {
|
||||
await Promise.all(pendingEvaluations)
|
||||
}
|
||||
// Now loop over the env types and evaluate them properly
|
||||
for (let i = 0; i < envTypes.length; i++) {
|
||||
let { name, value, type } = envTypes[i]
|
||||
// If an env-var wants to lookup itself, delegate straight to the parent
|
||||
// https://github.com/node-red/node-red/issues/2099
|
||||
if (value === name) {
|
||||
value = `$parent.${name}`
|
||||
}
|
||||
if (evaluatedEnv.hasOwnProperty(value)) {
|
||||
value = evaluatedEnv[value]
|
||||
} else {
|
||||
value = redUtil.evaluateNodeProperty(value, type, {_flow: {
|
||||
// Provide a hook so when it tries to look up a flow setting,
|
||||
// we can insert the just-evaluated value which hasn't yet
|
||||
// been set on the flow object - otherwise delegate up to the flow
|
||||
getSetting: function(name) {
|
||||
if (evaluatedEnv.hasOwnProperty(name)){
|
||||
return evaluatedEnv[name]
|
||||
}
|
||||
return flow.getSetting(name)
|
||||
}
|
||||
}}, null, null);
|
||||
}
|
||||
if (typeof value === 'object' && !value.__clone__) {
|
||||
value = { value: value, __clone__: true}
|
||||
}
|
||||
evaluatedEnv[name] = value
|
||||
|
||||
}
|
||||
// console.log(evaluatedEnv)
|
||||
|
||||
return evaluatedEnv
|
||||
}
|
||||
|
||||
/**
|
||||
* 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
|
||||
*/
|
||||
async function createNode(flow,config) {
|
||||
var newNode = null;
|
||||
var type = config.type;
|
||||
try {
|
||||
var nodeTypeConstructor = typeRegistry.get(type);
|
||||
if (typeof nodeTypeConstructor === "function") {
|
||||
var conf = jsonClone(config);
|
||||
delete conf.credentials;
|
||||
try {
|
||||
Object.defineProperty(conf,'_module', {value: typeRegistry.getNodeInfo(type), enumerable: false, writable: true })
|
||||
Object.defineProperty(conf,'_flow', {value: flow, enumerable: false, writable: true })
|
||||
Object.defineProperty(conf,'_path', {value: `${flow.path}/${config._alias||config.id}`, enumerable: false, writable: true })
|
||||
|
||||
for (var p in conf) {
|
||||
if (conf.hasOwnProperty(p)) {
|
||||
mapEnvVarProperties(conf,p,flow,conf);
|
||||
}
|
||||
}
|
||||
newNode = new nodeTypeConstructor(conf);
|
||||
} catch (err) {
|
||||
Log.log({
|
||||
level: Log.ERROR,
|
||||
id:conf.id,
|
||||
type: type,
|
||||
msg: err
|
||||
});
|
||||
}
|
||||
} else if (nodeTypeConstructor) {
|
||||
// console.log(nodeTypeConstructor)
|
||||
var subflowConfig = parseConfig([nodeTypeConstructor.subflow].concat(nodeTypeConstructor.subflow.flow));
|
||||
var subflowInstanceConfig = subflowConfig.subflows[nodeTypeConstructor.subflow.id];
|
||||
delete subflowConfig.subflows[nodeTypeConstructor.subflow.id];
|
||||
subflowInstanceConfig.subflows = subflowConfig.subflows;
|
||||
var instanceConfig = jsonClone(config);
|
||||
instanceConfig.env = jsonClone(nodeTypeConstructor.subflow.env);
|
||||
|
||||
instanceConfig.env = nodeTypeConstructor.subflow.env.map(nodeProp => {
|
||||
var nodePropType;
|
||||
var nodePropValue = config[nodeProp.name];
|
||||
if (nodeProp.type === "cred") {
|
||||
nodePropType = "cred";
|
||||
} else {
|
||||
switch(typeof config[nodeProp.name]) {
|
||||
case "string": nodePropType = "str"; break;
|
||||
case "number": nodePropType = "num"; break;
|
||||
case "boolean": nodePropType = "bool"; nodePropValue == nodeProp?"true":"false"; break;
|
||||
default:
|
||||
nodePropType = config[nodeProp.name].type;
|
||||
nodePropValue = config[nodeProp.name].value;
|
||||
}
|
||||
}
|
||||
return {
|
||||
name: nodeProp.name,
|
||||
type: nodePropType,
|
||||
value: nodePropValue
|
||||
}
|
||||
})
|
||||
|
||||
var subflow = require("./Subflow").createModuleInstance(
|
||||
nodeTypeConstructor.type,
|
||||
flow,
|
||||
flow.global,
|
||||
subflowInstanceConfig,
|
||||
instanceConfig
|
||||
);
|
||||
// Register this subflow as an instance node of the parent flow.
|
||||
// This allows nodes inside the subflow to get ahold of each other
|
||||
// such as a node accessing its config node
|
||||
flow.subflowInstanceNodes[config.id] = subflow
|
||||
await subflow.start();
|
||||
return subflow.node;
|
||||
}
|
||||
} catch(err) {
|
||||
Log.error(err);
|
||||
}
|
||||
return newNode;
|
||||
}
|
||||
|
||||
function parseConfig(config) {
|
||||
var flow = {};
|
||||
flow.allNodes = {};
|
||||
flow.subflows = {};
|
||||
flow.configs = {};
|
||||
flow.flows = {};
|
||||
flow.missingTypes = [];
|
||||
|
||||
config.forEach(function (n) {
|
||||
flow.allNodes[n.id] = jsonClone(n);
|
||||
if (n.type === 'tab') {
|
||||
flow.flows[n.id] = n;
|
||||
flow.flows[n.id].subflows = {};
|
||||
flow.flows[n.id].configs = {};
|
||||
flow.flows[n.id].nodes = {};
|
||||
flow.flows[n.id].groups = {};
|
||||
} else if (n.type === 'subflow') {
|
||||
flow.subflows[n.id] = n;
|
||||
flow.subflows[n.id].configs = {};
|
||||
flow.subflows[n.id].nodes = {};
|
||||
flow.subflows[n.id].groups = {};
|
||||
flow.subflows[n.id].instances = [];
|
||||
}
|
||||
});
|
||||
|
||||
var linkWires = {};
|
||||
var linkOutNodes = [];
|
||||
config.forEach(function (n) {
|
||||
if (n.type !== 'subflow' && n.type !== 'tab' && n.type !== 'group') {
|
||||
var subflowDetails = subflowInstanceRE.exec(n.type);
|
||||
|
||||
if ((subflowDetails && !flow.subflows[subflowDetails[1]]) || (!subflowDetails && !typeRegistry.get(n.type))) {
|
||||
if (flow.missingTypes.indexOf(n.type) === -1) {
|
||||
flow.missingTypes.push(n.type);
|
||||
}
|
||||
}
|
||||
var container = null;
|
||||
if (flow.flows[n.z]) {
|
||||
container = flow.flows[n.z];
|
||||
} else if (flow.subflows[n.z]) {
|
||||
container = flow.subflows[n.z];
|
||||
}
|
||||
if (n.hasOwnProperty('x') && n.hasOwnProperty('y')) {
|
||||
if (subflowDetails) {
|
||||
var subflowType = subflowDetails[1]
|
||||
n.subflow = subflowType;
|
||||
if (flow.subflows[subflowType]) {
|
||||
flow.subflows[subflowType].instances.push(n)
|
||||
}
|
||||
}
|
||||
if (container) {
|
||||
container.nodes[n.id] = n;
|
||||
}
|
||||
} else {
|
||||
if (container) {
|
||||
container.configs[n.id] = n;
|
||||
} else {
|
||||
flow.configs[n.id] = n;
|
||||
flow.configs[n.id]._users = [];
|
||||
}
|
||||
}
|
||||
if (n.type === 'link in' && n.links) {
|
||||
// Ensure wires are present in corresponding link out nodes
|
||||
n.links.forEach(function (id) {
|
||||
linkWires[id] = linkWires[id] || {};
|
||||
linkWires[id][n.id] = true;
|
||||
})
|
||||
} else if (n.type === 'link out' && n.links) {
|
||||
linkWires[n.id] = linkWires[n.id] || {};
|
||||
n.links.forEach(function (id) {
|
||||
linkWires[n.id][id] = true;
|
||||
})
|
||||
linkOutNodes.push(n);
|
||||
}
|
||||
} else if (n.type === 'group') {
|
||||
const parentContainer = flow.flows[n.z] || flow.subflows[n.z]
|
||||
if (parentContainer) {
|
||||
parentContainer.groups[n.id] = n
|
||||
}
|
||||
}
|
||||
});
|
||||
linkOutNodes.forEach(function (n) {
|
||||
var links = linkWires[n.id];
|
||||
var targets = Object.keys(links);
|
||||
n.wires = [targets];
|
||||
});
|
||||
|
||||
|
||||
var addedTabs = {};
|
||||
config.forEach(function (n) {
|
||||
if (n.type !== 'subflow' && n.type !== 'tab' && n.type !== 'group') {
|
||||
for (var prop in n) {
|
||||
if (n.hasOwnProperty(prop) && prop !== 'id' && prop !== 'wires' && prop !== 'type' && prop !== '_users' && flow.configs.hasOwnProperty(n[prop])) {
|
||||
// This property references a global config node
|
||||
flow.configs[n[prop]]._users.push(n.id)
|
||||
}
|
||||
}
|
||||
if (n.z && !flow.subflows[n.z]) {
|
||||
|
||||
if (!flow.flows[n.z]) {
|
||||
flow.flows[n.z] = { type: 'tab', id: n.z };
|
||||
flow.flows[n.z].subflows = {};
|
||||
flow.flows[n.z].configs = {};
|
||||
flow.flows[n.z].nodes = {};
|
||||
addedTabs[n.z] = flow.flows[n.z];
|
||||
}
|
||||
if (addedTabs[n.z]) {
|
||||
if (n.hasOwnProperty('x') && n.hasOwnProperty('y')) {
|
||||
addedTabs[n.z].nodes[n.id] = n;
|
||||
} else {
|
||||
addedTabs[n.z].configs[n.id] = n;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
return flow;
|
||||
}
|
||||
function getEnvVar(k) {
|
||||
if (!envVarExcludes[k]) {
|
||||
return process.env[k];
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
function diffConfigs(oldConfig, newConfig) {
|
||||
var id;
|
||||
var node;
|
||||
var nn;
|
||||
var wires;
|
||||
var j,k;
|
||||
|
||||
if (!oldConfig) {
|
||||
oldConfig = {
|
||||
flows:{},
|
||||
allNodes:{}
|
||||
}
|
||||
}
|
||||
var changedSubflows = {};
|
||||
|
||||
var added = {};
|
||||
var removed = {};
|
||||
var changed = {};
|
||||
var flowChanged = {};
|
||||
var wiringChanged = {};
|
||||
var globalConfigChanged = false;
|
||||
var linkMap = {};
|
||||
var allNestedGroups = []
|
||||
|
||||
// Look for tabs that have been removed
|
||||
for (id in oldConfig.flows) {
|
||||
if (oldConfig.flows.hasOwnProperty(id) && (!newConfig.flows.hasOwnProperty(id))) {
|
||||
removed[id] = oldConfig.allNodes[id];
|
||||
}
|
||||
}
|
||||
|
||||
// Look for tabs that have been disabled
|
||||
for (id in oldConfig.flows) {
|
||||
if (oldConfig.flows.hasOwnProperty(id) && newConfig.flows.hasOwnProperty(id)) {
|
||||
var originalState = oldConfig.flows[id].disabled||false;
|
||||
var newState = newConfig.flows[id].disabled||false;
|
||||
if (originalState !== newState) {
|
||||
if (originalState) {
|
||||
added[id] = oldConfig.allNodes[id];
|
||||
} else {
|
||||
removed[id] = oldConfig.allNodes[id];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (id in oldConfig.allNodes) {
|
||||
if (oldConfig.allNodes.hasOwnProperty(id)) {
|
||||
node = oldConfig.allNodes[id];
|
||||
if (node.type !== 'tab') {
|
||||
// build the map of what this node was previously wired to
|
||||
if (node.wires) {
|
||||
linkMap[node.id] = linkMap[node.id] || [];
|
||||
for (j=0;j<node.wires.length;j++) {
|
||||
wires = node.wires[j];
|
||||
for (k=0;k<wires.length;k++) {
|
||||
linkMap[node.id].push(wires[k]);
|
||||
nn = oldConfig.allNodes[wires[k]];
|
||||
if (nn) {
|
||||
linkMap[nn.id] = linkMap[nn.id] || [];
|
||||
linkMap[nn.id].push(node.id);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// This node has been removed or its flow disabled
|
||||
if (removed[node.z] || !newConfig.allNodes.hasOwnProperty(id)) {
|
||||
removed[id] = node;
|
||||
// Mark the container as changed
|
||||
if (!removed[node.z] && newConfig.allNodes[removed[id].z]) {
|
||||
changed[removed[id].z] = newConfig.allNodes[removed[id].z];
|
||||
if (changed[removed[id].z].type === "subflow") {
|
||||
changedSubflows[removed[id].z] = changed[removed[id].z];
|
||||
//delete removed[id];
|
||||
}
|
||||
}
|
||||
} else {
|
||||
if (added[node.z]) {
|
||||
added[id] = node;
|
||||
} else {
|
||||
var currentState = node.d;
|
||||
var newState = newConfig.allNodes[id].d;
|
||||
if (!currentState && newState) {
|
||||
removed[id] = node;
|
||||
}
|
||||
// This node has a material configuration change
|
||||
if (diffNodes(node,newConfig.allNodes[id]) || newConfig.allNodes[id].credentials) {
|
||||
changed[id] = newConfig.allNodes[id];
|
||||
if (changed[id].type === "subflow") {
|
||||
changedSubflows[id] = changed[id];
|
||||
}
|
||||
// Mark the container as changed
|
||||
if (newConfig.allNodes[changed[id].z]) {
|
||||
changed[changed[id].z] = newConfig.allNodes[changed[id].z];
|
||||
if (changed[changed[id].z].type === "subflow") {
|
||||
changedSubflows[changed[id].z] = changed[changed[id].z];
|
||||
delete changed[id];
|
||||
}
|
||||
}
|
||||
if (newConfig.allNodes[id].type === 'global-config') {
|
||||
globalConfigChanged = true
|
||||
}
|
||||
}
|
||||
// This node's wiring has changed
|
||||
if (!redUtil.compareObjects(node.wires,newConfig.allNodes[id].wires)) {
|
||||
wiringChanged[id] = newConfig.allNodes[id];
|
||||
// Mark the container as changed
|
||||
if (newConfig.allNodes[wiringChanged[id].z]) {
|
||||
changed[wiringChanged[id].z] = newConfig.allNodes[wiringChanged[id].z];
|
||||
if (changed[wiringChanged[id].z].type === "subflow") {
|
||||
changedSubflows[wiringChanged[id].z] = changed[wiringChanged[id].z];
|
||||
delete wiringChanged[id];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if (!removed[id]) {
|
||||
if (JSON.stringify(node.env) !== JSON.stringify(newConfig.allNodes[id].env)) {
|
||||
flowChanged[id] = newConfig.allNodes[id];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// Look for added nodes
|
||||
for (id in newConfig.allNodes) {
|
||||
if (newConfig.allNodes.hasOwnProperty(id)) {
|
||||
node = newConfig.allNodes[id];
|
||||
if (node.type === 'group') {
|
||||
if (node.g) {
|
||||
allNestedGroups.push(node)
|
||||
}
|
||||
if (changed[node.id]) {
|
||||
if (node.nodes) {
|
||||
node.nodes.forEach(nid => {
|
||||
if (!changed[nid]) {
|
||||
changed[nid] = true
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
// build the map of what this node is now wired to
|
||||
if (node.wires) {
|
||||
linkMap[node.id] = linkMap[node.id] || [];
|
||||
for (j=0;j<node.wires.length;j++) {
|
||||
wires = node.wires[j];
|
||||
for (k=0;k<wires.length;k++) {
|
||||
if (linkMap[node.id].indexOf(wires[k]) === -1) {
|
||||
linkMap[node.id].push(wires[k]);
|
||||
}
|
||||
nn = newConfig.allNodes[wires[k]];
|
||||
if (nn) {
|
||||
linkMap[nn.id] = linkMap[nn.id] || [];
|
||||
if (linkMap[nn.id].indexOf(node.id) === -1) {
|
||||
linkMap[nn.id].push(node.id);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// This node has been added
|
||||
if (!oldConfig.allNodes.hasOwnProperty(id)) {
|
||||
added[id] = node;
|
||||
// Mark the container as changed
|
||||
if (newConfig.allNodes[added[id].z]) {
|
||||
changed[added[id].z] = newConfig.allNodes[added[id].z];
|
||||
if (changed[added[id].z].type === "subflow") {
|
||||
changedSubflows[added[id].z] = changed[added[id].z];
|
||||
delete added[id];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
var madeChange;
|
||||
// Loop through the nodes looking for references to changed config nodes
|
||||
// Repeat the loop if anything is marked as changed as it may need to be
|
||||
// propagated to parent nodes.
|
||||
// TODO: looping through all nodes every time is a bit inefficient - could be more targeted
|
||||
do {
|
||||
madeChange = false;
|
||||
for (id in newConfig.allNodes) {
|
||||
if (newConfig.allNodes.hasOwnProperty(id)) {
|
||||
node = newConfig.allNodes[id];
|
||||
for (var prop in node) {
|
||||
if (node.hasOwnProperty(prop) && prop != "z" && prop != "id" && prop != "wires") {
|
||||
// This node has a property that references a changed/removed node
|
||||
// Assume it is a config node change and mark this node as
|
||||
// changed.
|
||||
|
||||
var changeOrigin = changed[node[prop]];
|
||||
if (changeOrigin || removed[node[prop]]) {
|
||||
if (!changed[node.id]) {
|
||||
if (changeOrigin &&
|
||||
(prop === "g") &&
|
||||
(changeOrigin.type === "group")) {
|
||||
var oldNode = oldConfig.allNodes[node.id];
|
||||
// ignore change of group node
|
||||
// if group of this node not changed
|
||||
if (oldNode &&
|
||||
(node.g === oldNode.g)) {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
madeChange = true;
|
||||
changed[node.id] = node;
|
||||
// This node exists within subflow template
|
||||
// Mark the template as having changed
|
||||
if (newConfig.allNodes[node.z]) {
|
||||
changed[node.z] = newConfig.allNodes[node.z];
|
||||
if (changed[node.z].type === "subflow") {
|
||||
changedSubflows[node.z] = changed[node.z];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} while (madeChange===true)
|
||||
|
||||
// Find any nodes that exist on a subflow template and remove from changed
|
||||
// list as the parent subflow will now be marked as containing a change
|
||||
for (id in newConfig.allNodes) {
|
||||
if (newConfig.allNodes.hasOwnProperty(id)) {
|
||||
node = newConfig.allNodes[id];
|
||||
if (newConfig.allNodes[node.z] && newConfig.allNodes[node.z].type === "subflow") {
|
||||
delete changed[node.id];
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Recursively mark all children of changed groups as changed
|
||||
do {
|
||||
madeChange = false
|
||||
for (let i = 0; i < allNestedGroups.length; i++) {
|
||||
const group = allNestedGroups[i]
|
||||
if (!changed[group.id] && group.g && changed[group.g]) {
|
||||
changed[group.id] = true
|
||||
madeChange = true
|
||||
}
|
||||
if (changed[group.id] && group.nodes) {
|
||||
group.nodes.forEach(nid => {
|
||||
if (!changed[nid]) {
|
||||
changed[nid] = true
|
||||
madeChange = true
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
} while(madeChange)
|
||||
|
||||
// Recursively mark all instances of changed subflows as changed
|
||||
var changedSubflowStack = Object.keys(changedSubflows);
|
||||
while (changedSubflowStack.length > 0) {
|
||||
var subflowId = changedSubflowStack.pop();
|
||||
for (id in newConfig.allNodes) {
|
||||
if (newConfig.allNodes.hasOwnProperty(id)) {
|
||||
node = newConfig.allNodes[id];
|
||||
if (node.type === 'subflow:'+subflowId) {
|
||||
if (!changed[node.id]) {
|
||||
changed[node.id] = node;
|
||||
if (!changed[changed[node.id].z] && newConfig.allNodes[changed[node.id].z]) {
|
||||
changed[changed[node.id].z] = newConfig.allNodes[changed[node.id].z];
|
||||
if (newConfig.allNodes[changed[node.id].z].type === "subflow") {
|
||||
// This subflow instance is inside a subflow. Add the
|
||||
// containing subflow to the stack to mark
|
||||
changedSubflowStack.push(changed[node.id].z);
|
||||
delete changed[node.id];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
var diff = {
|
||||
added:Object.keys(added),
|
||||
changed:Object.keys(changed),
|
||||
removed:Object.keys(removed),
|
||||
rewired:Object.keys(wiringChanged),
|
||||
linked:[],
|
||||
flowChanged: Object.keys(flowChanged),
|
||||
globalConfigChanged
|
||||
}
|
||||
|
||||
// Traverse the links of all modified nodes to mark the connected nodes
|
||||
var modifiedNodes = diff.added.concat(diff.changed).concat(diff.removed).concat(diff.rewired);
|
||||
var visited = {};
|
||||
while (modifiedNodes.length > 0) {
|
||||
node = modifiedNodes.pop();
|
||||
if (!visited[node]) {
|
||||
visited[node] = true;
|
||||
if (linkMap[node]) {
|
||||
if (!changed[node] && !added[node] && !removed[node] && !wiringChanged[node]) {
|
||||
diff.linked.push(node);
|
||||
}
|
||||
modifiedNodes = modifiedNodes.concat(linkMap[node]);
|
||||
}
|
||||
}
|
||||
}
|
||||
// console.log(diff);
|
||||
// for (id in newConfig.allNodes) {
|
||||
// if (added[id] || changed[id] || wiringChanged[id] || diff.linked.indexOf(id)!==-1) {
|
||||
// console.log(
|
||||
// (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||""
|
||||
// );
|
||||
// }
|
||||
// }
|
||||
// for (id in removed) {
|
||||
// console.log(
|
||||
// "- "+(diff.linked.indexOf(id)!==-1?"~":" "),
|
||||
// id,
|
||||
// oldConfig.allNodes[id].type,
|
||||
// oldConfig.allNodes[id].name||oldConfig.allNodes[id].label||""
|
||||
// );
|
||||
// }
|
||||
|
||||
return diff;
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
init,
|
||||
createNode,
|
||||
parseConfig,
|
||||
diffConfigs,
|
||||
diffNodes,
|
||||
getEnvVar,
|
||||
mapEnvVarProperties,
|
||||
evaluateEnvProperties
|
||||
}
|
||||
Reference in New Issue
Block a user