mirror of
https://github.com/node-red/node-red.git
synced 2025-03-01 10:36:34 +00:00
Flow Engine refactor
Each flow/tab now exists as its own logical object. This is the ground work for allowing flows to be added/removed/updated independently.
This commit is contained in:
Nick O'Leary
467
red/nodes/flows/Flow.js
Normal file
467
red/nodes/flows/Flow.js
Normal file
@@ -0,0 +1,467 @@
|
||||
/**
|
||||
* Copyright 2015 IBM Corp.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
**/
|
||||
|
||||
var when = require("when");
|
||||
var clone = require("clone");
|
||||
var typeRegistry = require("../registry");
|
||||
var Log = require("../../log");
|
||||
var redUtil = require("../../util");
|
||||
var flowUtil = require("./util");
|
||||
|
||||
function Flow(global,flow) {
|
||||
if (typeof flow === 'undefined') {
|
||||
flow = global;
|
||||
}
|
||||
var activeNodes = {};
|
||||
var subflowInstanceNodes = {};
|
||||
var catchNodeMap = {};
|
||||
var statusNodeMap = {};
|
||||
|
||||
this.start = function(diff) {
|
||||
var node;
|
||||
var id;
|
||||
catchNodeMap = {};
|
||||
statusNodeMap = {};
|
||||
for (id in flow.configs) {
|
||||
if (flow.configs.hasOwnProperty(id)) {
|
||||
node = flow.configs[id];
|
||||
if (!activeNodes[id]) {
|
||||
activeNodes[id] = createNode(node.type,node);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (diff) {
|
||||
for (var j=0;j<diff.rewired.length;j++) {
|
||||
var rewireNode = activeNodes[diff.rewired[j]];
|
||||
if (rewireNode) {
|
||||
rewireNode.updateWires(flow.nodes[rewireNode.id].wires);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (id in flow.nodes) {
|
||||
if (flow.nodes.hasOwnProperty(id)) {
|
||||
node = flow.nodes[id];
|
||||
if (!node.subflow) {
|
||||
if (!activeNodes[id]) {
|
||||
activeNodes[id] = createNode(node.type,node);
|
||||
}
|
||||
} else {
|
||||
if (!subflowInstanceNodes[id]) {
|
||||
try {
|
||||
var nodes = createSubflow(flow.subflows[node.subflow]||global.subflows[node.subflow],node,flow.subflows,global.subflows,activeNodes);
|
||||
subflowInstanceNodes[id] = nodes.map(function(n) { return n.id});
|
||||
for (var i=0;i<nodes.length;i++) {
|
||||
activeNodes[nodes[i].id] = nodes[i];
|
||||
}
|
||||
} catch(err) {
|
||||
console.log(err.stack)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (id in activeNodes) {
|
||||
if (activeNodes.hasOwnProperty(id)) {
|
||||
node = activeNodes[id];
|
||||
if (node.type === "catch") {
|
||||
catchNodeMap[node.z] = catchNodeMap[node.z] || [];
|
||||
catchNodeMap[node.z].push(node);
|
||||
} else if (node.type === "status") {
|
||||
statusNodeMap[node.z] = statusNodeMap[node.z] || [];
|
||||
statusNodeMap[node.z].push(node);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
this.stop = function(stopList) {
|
||||
return when.promise(function(resolve) {
|
||||
var i;
|
||||
if (stopList) {
|
||||
for (i=0;i<stopList.length;i++) {
|
||||
if (subflowInstanceNodes[stopList[i]]) {
|
||||
// The first in the list is the instance node we already
|
||||
// know about
|
||||
stopList = stopList.concat(subflowInstanceNodes[stopList[i]].slice(1))
|
||||
}
|
||||
}
|
||||
} else {
|
||||
stopList = Object.keys(activeNodes);
|
||||
}
|
||||
var promises = [];
|
||||
for (i=0;i<stopList.length;i++) {
|
||||
var node = activeNodes[stopList[i]];
|
||||
if (node) {
|
||||
delete activeNodes[stopList[i]];
|
||||
if (subflowInstanceNodes[stopList[i]]) {
|
||||
delete subflowInstanceNodes[stopList[i]];
|
||||
}
|
||||
try {
|
||||
var p = node.close();
|
||||
if (p) {
|
||||
promises.push(p);
|
||||
}
|
||||
} catch(err) {
|
||||
node.error(err);
|
||||
}
|
||||
}
|
||||
}
|
||||
when.settle(promises).then(function() {
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
this.update = function(_global,_flow) {
|
||||
global = _global;
|
||||
flow = _flow;
|
||||
}
|
||||
|
||||
this.getNode = function(id) {
|
||||
return activeNodes[id];
|
||||
}
|
||||
|
||||
this.getActiveNodes = function() {
|
||||
return activeNodes;
|
||||
}
|
||||
|
||||
this.handleStatus = function(node,statusMessage) {
|
||||
var targetStatusNodes = null;
|
||||
var reportingNode = node;
|
||||
var handled = false;
|
||||
while(reportingNode && !handled) {
|
||||
targetStatusNodes = statusNodeMap[reportingNode.z];
|
||||
if (targetStatusNodes) {
|
||||
targetStatusNodes.forEach(function(targetStatusNode) {
|
||||
if (targetStatusNode.scope && targetStatusNode.scope.indexOf(node.id) === -1) {
|
||||
return;
|
||||
}
|
||||
var message = {
|
||||
status: {
|
||||
text: "",
|
||||
source: {
|
||||
id: node.id,
|
||||
type: node.type,
|
||||
name: node.name
|
||||
}
|
||||
}
|
||||
};
|
||||
if (statusMessage.text) {
|
||||
message.status.text = statusMessage.text;
|
||||
}
|
||||
targetStatusNode.receive(message);
|
||||
handled = true;
|
||||
});
|
||||
}
|
||||
if (!handled) {
|
||||
reportingNode = activeNodes[reportingNode.z];
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
this.handleError = function(node,logMessage,msg) {
|
||||
var count = 1;
|
||||
if (msg && msg.hasOwnProperty("error")) {
|
||||
if (msg.error.hasOwnProperty("source")) {
|
||||
if (msg.error.source.id === node.id) {
|
||||
count = msg.error.source.count+1;
|
||||
if (count === 10) {
|
||||
node.warn(Log._("nodes.flow.error-loop"));
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
var targetCatchNodes = null;
|
||||
var throwingNode = node;
|
||||
var handled = false;
|
||||
while (throwingNode && !handled) {
|
||||
targetCatchNodes = catchNodeMap[throwingNode.z];
|
||||
if (targetCatchNodes) {
|
||||
targetCatchNodes.forEach(function(targetCatchNode) {
|
||||
if (targetCatchNode.scope && targetCatchNode.scope.indexOf(throwingNode.id) === -1) {
|
||||
return;
|
||||
}
|
||||
var errorMessage;
|
||||
if (msg) {
|
||||
errorMessage = redUtil.cloneMessage(msg);
|
||||
} else {
|
||||
errorMessage = {};
|
||||
}
|
||||
if (errorMessage.hasOwnProperty("error")) {
|
||||
errorMessage._error = errorMessage.error;
|
||||
}
|
||||
errorMessage.error = {
|
||||
message: logMessage.toString(),
|
||||
source: {
|
||||
id: node.id,
|
||||
type: node.type,
|
||||
count: count
|
||||
}
|
||||
};
|
||||
targetCatchNode.receive(errorMessage);
|
||||
handled = true;
|
||||
});
|
||||
}
|
||||
if (!handled) {
|
||||
throwingNode = activeNodes[throwingNode.z];
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
function getID() {
|
||||
return (1+Math.random()*4294967295).toString(16);
|
||||
}
|
||||
|
||||
var EnvVarPropertyRE = /^\$\((\S+)\)$/;
|
||||
|
||||
function mapEnvVarProperties(obj,prop) {
|
||||
if (Buffer.isBuffer(obj[prop])) {
|
||||
return;
|
||||
} else if (Array.isArray(obj[prop])) {
|
||||
for (var i=0;i<obj[prop].length;i++) {
|
||||
mapEnvVarProperties(obj[prop],i);
|
||||
}
|
||||
} else if (typeof obj[prop] === 'string') {
|
||||
var m;
|
||||
if ( (m = EnvVarPropertyRE.exec(obj[prop])) !== null) {
|
||||
if (process.env.hasOwnProperty(m[1])) {
|
||||
obj[prop] = process.env[m[1]];
|
||||
}
|
||||
}
|
||||
} else {
|
||||
for (var p in obj[prop]) {
|
||||
if (obj[prop].hasOwnProperty) {
|
||||
mapEnvVarProperties(obj[prop],p);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function createNode(type,config) {
|
||||
//console.log("CREATE",type,config.id);
|
||||
var nn = null;
|
||||
var nt = typeRegistry.get(type);
|
||||
if (nt) {
|
||||
var conf = clone(config);
|
||||
for (var p in conf) {
|
||||
if (conf.hasOwnProperty(p)) {
|
||||
mapEnvVarProperties(conf,p);
|
||||
}
|
||||
}
|
||||
try {
|
||||
nn = new nt(conf);
|
||||
}
|
||||
catch (err) {
|
||||
Log.log({
|
||||
level: Log.ERROR,
|
||||
id:conf.id,
|
||||
type: type,
|
||||
msg: err
|
||||
});
|
||||
}
|
||||
} else {
|
||||
Log.error(Log._("nodes.flow.unknown-type", {type:type}));
|
||||
}
|
||||
return nn;
|
||||
}
|
||||
|
||||
function createSubflow(sf,sfn,subflows,globalSubflows,activeNodes) {
|
||||
//console.log("CREATE SUBFLOW",sf.id,sfn.id);
|
||||
var nodes = [];
|
||||
var node_map = {};
|
||||
var newNodes = [];
|
||||
var node;
|
||||
var wires;
|
||||
var i,j,k;
|
||||
|
||||
var createNodeInSubflow = function(def) {
|
||||
node = clone(def);
|
||||
var nid = getID();
|
||||
node_map[node.id] = node;
|
||||
node._alias = node.id;
|
||||
node.id = nid;
|
||||
node.z = sfn.id;
|
||||
newNodes.push(node);
|
||||
}
|
||||
|
||||
// Clone all of the subflow node definitions and give them new IDs
|
||||
for (i in sf.configs) {
|
||||
if (sf.configs.hasOwnProperty(i)) {
|
||||
createNodeInSubflow(sf.configs[i]);
|
||||
}
|
||||
}
|
||||
// Clone all of the subflow node definitions and give them new IDs
|
||||
for (i in sf.nodes) {
|
||||
if (sf.nodes.hasOwnProperty(i)) {
|
||||
createNodeInSubflow(sf.nodes[i]);
|
||||
}
|
||||
}
|
||||
|
||||
// Look for any catch/status nodes and update their scope ids
|
||||
// Update all subflow interior wiring to reflect new node IDs
|
||||
for (i=0;i<newNodes.length;i++) {
|
||||
node = newNodes[i];
|
||||
if (node.wires) {
|
||||
var outputs = node.wires;
|
||||
for (j=0;j<outputs.length;j++) {
|
||||
wires = outputs[j];
|
||||
for (k=0;k<wires.length;k++) {
|
||||
outputs[j][k] = node_map[outputs[j][k]].id
|
||||
}
|
||||
}
|
||||
if ((node.type === 'catch' || node.type === 'status') && node.scope) {
|
||||
node.scope = node.scope.map(function(id) {
|
||||
return node_map[id]?node_map[id].id:""
|
||||
})
|
||||
} else {
|
||||
for (var prop in node) {
|
||||
if (node.hasOwnProperty(prop) && prop !== '_alias') {
|
||||
if (node_map[node[prop]]) {
|
||||
//console.log("Mapped",node.type,node.id,prop,node_map[node[prop]].id);
|
||||
node[prop] = node_map[node[prop]].id;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Create a subflow node to accept inbound messages and route appropriately
|
||||
var Node = require("../Node");
|
||||
var subflowInstance = {
|
||||
id: sfn.id,
|
||||
type: sfn.type,
|
||||
z: sfn.z,
|
||||
name: sfn.name,
|
||||
wires: []
|
||||
}
|
||||
if (sf.in) {
|
||||
subflowInstance.wires = sf.in.map(function(n) { return n.wires.map(function(w) { return node_map[w.id].id;})})
|
||||
subflowInstance._originalWires = clone(subflowInstance.wires);
|
||||
}
|
||||
var subflowNode = new Node(subflowInstance);
|
||||
|
||||
subflowNode.on("input", function(msg) { this.send(msg);});
|
||||
|
||||
|
||||
subflowNode._updateWires = subflowNode.updateWires;
|
||||
|
||||
subflowNode.updateWires = function(newWires) {
|
||||
// Wire the subflow outputs
|
||||
if (sf.out) {
|
||||
var node,wires,i,j;
|
||||
// Restore the original wiring to the internal nodes
|
||||
|
||||
subflowInstance.wires = clone(subflowInstance._originalWires);
|
||||
|
||||
for (i=0;i<sf.out.length;i++) {
|
||||
wires = sf.out[i].wires;
|
||||
for (j=0;j<wires.length;j++) {
|
||||
if (wires[j].id != sf.id) {
|
||||
node = node_map[wires[j].id];
|
||||
if (node._originalWires) {
|
||||
node.wires = clone(node._originalWires);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
var modifiedNodes = {};
|
||||
var subflowInstanceModified = false;
|
||||
|
||||
for (i=0;i<sf.out.length;i++) {
|
||||
wires = sf.out[i].wires;
|
||||
for (j=0;j<wires.length;j++) {
|
||||
if (wires[j].id === sf.id) {
|
||||
subflowInstance.wires[wires[j].port] = subflowInstance.wires[wires[j].port].concat(newWires[i]);
|
||||
subflowInstanceModified = true;
|
||||
} else {
|
||||
node = node_map[wires[j].id];
|
||||
node.wires[wires[j].port] = node.wires[wires[j].port].concat(newWires[i]);
|
||||
modifiedNodes[node.id] = node;
|
||||
}
|
||||
}
|
||||
}
|
||||
Object.keys(modifiedNodes).forEach(function(id) {
|
||||
var node = modifiedNodes[id];
|
||||
subflowNode.instanceNodes[id].updateWires(node.wires);
|
||||
});
|
||||
if (subflowInstanceModified) {
|
||||
subflowNode._updateWires(subflowInstance.wires);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
nodes.push(subflowNode);
|
||||
|
||||
// Wire the subflow outputs
|
||||
if (sf.out) {
|
||||
var modifiedNodes = {};
|
||||
for (i=0;i<sf.out.length;i++) {
|
||||
wires = sf.out[i].wires;
|
||||
for (j=0;j<wires.length;j++) {
|
||||
if (wires[j].id === sf.id) {
|
||||
// A subflow input wired straight to a subflow output
|
||||
subflowInstance.wires[wires[j].port] = subflowInstance.wires[wires[j].port].concat(sfn.wires[i])
|
||||
subflowNode._updateWires(subflowInstance.wires);
|
||||
} else {
|
||||
node = node_map[wires[j].id];
|
||||
modifiedNodes[node.id] = node;
|
||||
if (!node._originalWires) {
|
||||
node._originalWires = clone(node.wires);
|
||||
}
|
||||
node.wires[wires[j].port] = (node.wires[wires[j].port]||[]).concat(sfn.wires[i]);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Instantiate the nodes
|
||||
for (i=0;i<newNodes.length;i++) {
|
||||
node = newNodes[i];
|
||||
var type = node.type;
|
||||
|
||||
var m = /^subflow:(.+)$/.exec(type);
|
||||
if (!m) {
|
||||
var newNode = createNode(type,node);
|
||||
activeNodes[node.id] = newNode;
|
||||
nodes.push(newNode);
|
||||
} else {
|
||||
var subflowId = m[1];
|
||||
nodes = nodes.concat(createSubflow(subflows[subflowId]||globalSubflows[subflowId],node,subflows,globalSubflows,activeNodes));
|
||||
}
|
||||
}
|
||||
|
||||
subflowNode.instanceNodes = {};
|
||||
|
||||
nodes.forEach(function(node) {
|
||||
subflowNode.instanceNodes[node.id] = node;
|
||||
});
|
||||
return nodes;
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
create: function(global,conf) {
|
||||
return new Flow(global,conf);
|
||||
}
|
||||
}
|
||||
318
red/nodes/flows/index.js
Normal file
318
red/nodes/flows/index.js
Normal file
@@ -0,0 +1,318 @@
|
||||
/**
|
||||
* Copyright 2014, 2015 IBM Corp.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
**/
|
||||
|
||||
var clone = require("clone");
|
||||
var when = require("when");
|
||||
|
||||
var Flow = require('./Flow');
|
||||
|
||||
var typeRegistry = require("../registry");
|
||||
var credentials = require("../credentials");
|
||||
|
||||
var flowUtil = require("./util");
|
||||
var log = require("../../log");
|
||||
var events = require("../../events");
|
||||
var redUtil = require("../../util");
|
||||
var deprecated = require("../registry/deprecated");
|
||||
|
||||
var storage = null;
|
||||
var settings = null;
|
||||
|
||||
var activeConfig = null;
|
||||
var activeFlowConfig = null;
|
||||
|
||||
var activeFlows = {};
|
||||
var started = false;
|
||||
|
||||
var activeNodesToFlow = {};
|
||||
|
||||
function init(_settings, _storage) {
|
||||
settings = _settings;
|
||||
storage = _storage;
|
||||
started = false;
|
||||
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) {
|
||||
start();
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
function load() {
|
||||
return storage.getFlows().then(function(flows) {
|
||||
return credentials.load().then(function() {
|
||||
return setConfig(flows,"load");
|
||||
});
|
||||
}).otherwise(function(err) {
|
||||
log.warn(log._("nodes.flows.error",{message:err.toString()}));
|
||||
console.log(err.stack);
|
||||
});
|
||||
}
|
||||
|
||||
function setConfig(config,type) {
|
||||
type = type||"full";
|
||||
|
||||
var credentialsChanged = false;
|
||||
var credentialSavePromise = null;
|
||||
var configSavePromise = null;
|
||||
|
||||
var cleanConfig = clone(config);
|
||||
cleanConfig.forEach(function(node) {
|
||||
if (node.credentials) {
|
||||
credentials.extract(node);
|
||||
credentialsChanged = true;
|
||||
}
|
||||
});
|
||||
|
||||
if (credentialsChanged) {
|
||||
credentialSavePromise = credentials.save();
|
||||
} else {
|
||||
credentialSavePromise = when.resolve();
|
||||
}
|
||||
if (type === 'load') {
|
||||
configSavePromise = credentialSavePromise;
|
||||
type = 'full';
|
||||
} else {
|
||||
configSavePromise = credentialSavePromise.then(function() {
|
||||
return storage.saveFlows(cleanConfig);
|
||||
});
|
||||
}
|
||||
|
||||
return configSavePromise
|
||||
.then(function() {
|
||||
var diff;
|
||||
activeConfig = cleanConfig;
|
||||
if (type === 'full') {
|
||||
activeFlowConfig = flowUtil.parseConfig(clone(config));
|
||||
} else {
|
||||
var newConfig = flowUtil.parseConfig(clone(config));
|
||||
diff = flowUtil.diffConfigs(activeFlowConfig,newConfig);
|
||||
activeFlowConfig = newConfig;
|
||||
}
|
||||
credentials.clean(activeConfig).then(function() {
|
||||
if (started) {
|
||||
return stop(type,diff).then(function() {
|
||||
start(type,diff);
|
||||
}).otherwise(function(err) {
|
||||
console.log(err);
|
||||
})
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
function getNode(id) {
|
||||
var node;
|
||||
if (activeNodesToFlow[id]) {
|
||||
return activeFlows[activeNodesToFlow[id]].getNode(id);
|
||||
}
|
||||
for (var flowId in activeFlows) {
|
||||
if (activeFlows.hasOwnProperty(flowId)) {
|
||||
node = activeFlows[flowId].getNode(id);
|
||||
if (node) {
|
||||
return node;
|
||||
}
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function eachNode(cb) {
|
||||
for (var id in activeFlowConfig.nodes) {
|
||||
if (activeFlowConfig.nodes.hasOwnProperty(id)) {
|
||||
cb(activeFlowConfig.nodes[id]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function getConfig() {
|
||||
return activeConfig;
|
||||
}
|
||||
|
||||
function handleError(node,logMessage,msg) {
|
||||
if (activeFlows[node.z]) {
|
||||
activeFlows[node.z].handleError(node,logMessage,msg);
|
||||
} else if (activeNodesToFlow[node.z]) {
|
||||
activeFlows[activeNodesToFlow[node.z]].handleError(node,logMessage,msg);
|
||||
}
|
||||
}
|
||||
|
||||
function handleStatus(node,statusMessage) {
|
||||
if (activeFlows[node.z]) {
|
||||
activeFlows[node.z].handleStatus(node,statusMessage);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
function start(type,diff) {
|
||||
type = type||"full";
|
||||
started = true;
|
||||
var i;
|
||||
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);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (diff) {
|
||||
log.info(log._("nodes.flows.starting-modified-"+type));
|
||||
} else {
|
||||
log.info(log._("nodes.flows.starting-flows"));
|
||||
}
|
||||
var id;
|
||||
if (!diff) {
|
||||
activeFlows['_GLOBAL_'] = Flow.create(activeFlowConfig);
|
||||
for (id in activeFlowConfig.flows) {
|
||||
if (activeFlowConfig.flows.hasOwnProperty(id)) {
|
||||
activeFlows[id] = Flow.create(activeFlowConfig,activeFlowConfig.flows[id]);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
activeFlows['_GLOBAL_'].update(activeFlowConfig,activeFlowConfig);
|
||||
for (id in activeFlowConfig.flows) {
|
||||
if (activeFlowConfig.flows.hasOwnProperty(id)) {
|
||||
if (activeFlows[id]) {
|
||||
activeFlows[id].update(activeFlowConfig,activeFlowConfig.flows[id]);
|
||||
} else {
|
||||
activeFlows[id] = Flow.create(activeFlowConfig,activeFlowConfig.flows[id]);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (id in activeFlows) {
|
||||
if (activeFlows.hasOwnProperty(id)) {
|
||||
activeFlows[id].start(diff);
|
||||
var activeNodes = activeFlows[id].getActiveNodes();
|
||||
Object.keys(activeNodes).forEach(function(nid) {
|
||||
activeNodesToFlow[nid] = id;
|
||||
});
|
||||
}
|
||||
}
|
||||
events.emit("nodes-started");
|
||||
if (diff) {
|
||||
log.info(log._("nodes.flows.started-modified-"+type));
|
||||
} else {
|
||||
log.info(log._("nodes.flows.started-flows"));
|
||||
}
|
||||
}
|
||||
|
||||
function stop(type,diff) {
|
||||
type = type||"full";
|
||||
if (diff) {
|
||||
log.info(log._("nodes.flows.stopping-modified-"+type));
|
||||
} else {
|
||||
log.info(log._("nodes.flows.stopping-flows"));
|
||||
}
|
||||
started = false;
|
||||
var promises = [];
|
||||
var stopList;
|
||||
if (type === 'nodes') {
|
||||
stopList = diff.changed.concat(diff.removed);
|
||||
} else if (type === 'flows') {
|
||||
stopList = diff.changed.concat(diff.removed).concat(diff.linked);
|
||||
}
|
||||
for (var id in activeFlows) {
|
||||
if (activeFlows.hasOwnProperty(id)) {
|
||||
promises = promises.concat(activeFlows[id].stop(stopList));
|
||||
if (!diff || diff.removed[id]) {
|
||||
delete activeFlows[id];
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return when.promise(function(resolve,reject) {
|
||||
when.settle(promises).then(function() {
|
||||
for (id in activeNodesToFlow) {
|
||||
if (activeNodesToFlow.hasOwnProperty(id)) {
|
||||
if (!activeFlows[activeNodesToFlow[id]]) {
|
||||
delete activeNodesToFlow[id];
|
||||
}
|
||||
}
|
||||
}
|
||||
if (diff) {
|
||||
log.info(log._("nodes.flows.stopped-modified-"+type));
|
||||
} else {
|
||||
log.info(log._("nodes.flows.stopped-flows"));
|
||||
}
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
module.exports = {
|
||||
init: init,
|
||||
|
||||
/**
|
||||
* Load the current flow configuration from storage
|
||||
* @return a promise for the loading of the config
|
||||
*/
|
||||
load: load,
|
||||
|
||||
get:getNode,
|
||||
eachNode: eachNode,
|
||||
|
||||
/**
|
||||
* Gets the current flow configuration
|
||||
*/
|
||||
getFlows: getConfig,
|
||||
|
||||
/**
|
||||
* 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: setConfig,
|
||||
|
||||
/**
|
||||
* Starts the current flow configuration
|
||||
*/
|
||||
startFlows: start,
|
||||
|
||||
/**
|
||||
* Stops the current flow configuration
|
||||
* @return a promise for the stopping of the flow
|
||||
*/
|
||||
stopFlows: stop,
|
||||
|
||||
|
||||
handleError: handleError,
|
||||
handleStatus: handleStatus
|
||||
};
|
||||
316
red/nodes/flows/util.js
Normal file
316
red/nodes/flows/util.js
Normal file
@@ -0,0 +1,316 @@
|
||||
/**
|
||||
* Copyright 2015 IBM Corp.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
**/
|
||||
var clone = require("clone");
|
||||
var redUtil = require("../../util");
|
||||
var subflowInstanceRE = /^subflow:(.+)$/;
|
||||
var typeRegistry = require("../registry");
|
||||
|
||||
function diffNodes(oldNode,newNode) {
|
||||
if (oldNode == null) {
|
||||
return true;
|
||||
}
|
||||
var oldKeys = Object.keys(oldNode).filter(function(p) { return p != "x" && p != "y" && p != "wires" });
|
||||
var newKeys = Object.keys(newNode).filter(function(p) { return p != "x" && p != "y" && p != "wires" });
|
||||
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;
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
|
||||
diffNodes: diffNodes,
|
||||
|
||||
parseConfig: function(config) {
|
||||
var flow = {};
|
||||
flow.allNodes = {};
|
||||
flow.subflows = {};
|
||||
flow.configs = {};
|
||||
flow.flows = {};
|
||||
flow.missingTypes = [];
|
||||
|
||||
config.forEach(function(n) {
|
||||
flow.allNodes[n.id] = clone(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 = {};
|
||||
}
|
||||
});
|
||||
|
||||
config.forEach(function(n) {
|
||||
if (n.type === 'subflow') {
|
||||
flow.subflows[n.id] = n;
|
||||
flow.subflows[n.id].configs = {};
|
||||
flow.subflows[n.id].nodes = {};
|
||||
flow.subflows[n.id].instances = [];
|
||||
}
|
||||
});
|
||||
|
||||
config.forEach(function(n) {
|
||||
if (n.type !== 'subflow' && n.type !== 'tab') {
|
||||
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);
|
||||
}
|
||||
} else {
|
||||
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;
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
return flow;
|
||||
},
|
||||
|
||||
diffConfigs: function(oldConfig, newConfig) {
|
||||
var id;
|
||||
var node;
|
||||
var nn;
|
||||
var wires;
|
||||
var j,k;
|
||||
|
||||
var changedSubflows = {};
|
||||
|
||||
var added = {};
|
||||
var removed = {};
|
||||
var changed = {};
|
||||
var wiringChanged = {};
|
||||
|
||||
var linkMap = {};
|
||||
|
||||
for (id in oldConfig.allNodes) {
|
||||
if (oldConfig.allNodes.hasOwnProperty(id)) {
|
||||
node = oldConfig.allNodes[id];
|
||||
// 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
|
||||
if (!newConfig.allNodes.hasOwnProperty(id)) {
|
||||
removed[id] = node;
|
||||
// Mark the container as changed
|
||||
if (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 {
|
||||
// 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];
|
||||
}
|
||||
}
|
||||
}
|
||||
// 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];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// Look for added nodes
|
||||
for (id in newConfig.allNodes) {
|
||||
if (newConfig.allNodes.hasOwnProperty(id)) {
|
||||
node = newConfig.allNodes[id];
|
||||
// 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];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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.
|
||||
if (changed[node[prop]] || removed[node[prop]]) {
|
||||
if (!changed[node.id]) {
|
||||
changed[node.id] = node;
|
||||
if (newConfig.allNodes[node.z]) {
|
||||
changed[node.z] = newConfig.allNodes[node.z];
|
||||
if (changed[node.z].type === "subflow") {
|
||||
changedSubflows[node.z] = changed[node.z];
|
||||
delete changed[node.id];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// 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:[]
|
||||
}
|
||||
|
||||
// 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]);
|
||||
}
|
||||
}
|
||||
}
|
||||
// for (id in newConfig.allNodes) {
|
||||
// console.log(
|
||||
// (added[id]?"+":(changed[id]?"!":" "))+(wiringChanged[id]?"w":" ")+(diff.linked.indexOf(id)!==-1?"~":" "),
|
||||
// id,
|
||||
// newConfig.allNodes[id].type,
|
||||
// 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;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user