node-red/packages/node_modules/@node-red/runtime/lib/nodes/flows/Flow.js

/**
 * 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.
 **/

var clone = require("clone");
var redUtil = require("@node-red/util").util;
var flowUtil = require("./util");
var events = require("../../events");

var Subflow;
var Log;

var nodeCloseTimeout = 15000;

/**
 * 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.activeNodes = {};
        this.subflowInstanceNodes = {};
        this.catchNodes = [];
        this.statusNodes = [];
    }

    /**
     * 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 a info-level message from this flow
     * @param  {[type]} msg [description]
     * @return {[type]}     [description]
     */
    log(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
        })
    }


    /**
     * 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]
     */
    start(diff) {
        this.trace("start "+this.TYPE);
        var node;
        var newNode;
        var id;
        this.catchNodes = [];
        this.statusNodes = [];

        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]) {
                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]]) {
                        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 = flowUtil.createNode(this,node);
                    if (newNode) {
                        this.activeNodes[id] = newNode;
                    }
                }
            }
        }

        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.subflow) {
                    if (!this.activeNodes[id]) {
                        newNode = 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;
                            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)
                        }
                    }
                }
            }
        }

        var activeCount = Object.keys(this.activeNodes).length;
        if (activeCount > 0) {
            this.trace("------------------|--------------|-----------------");
            this.trace(" id               | type         | alias");
            this.trace("------------------|--------------|-----------------");
        }
        // Build the map of catch/status nodes.
        for (id in this.activeNodes) {
            if (this.activeNodes.hasOwnProperty(id)) {
                node = this.activeNodes[id];
                this.trace(" "+id.padEnd(16)+" | "+node.type.padEnd(12)+" | "+(node._alias||""));
                if (node.type === "catch") {
                    this.catchNodes.push(node);
                } else if (node.type === "status") {
                    this.statusNodes.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);
        }
        // Convert the list to a map to avoid multiple scans of the list
        var removedMap = {};
        removedList = removedList || [];
        removedList.forEach(function(id) {
            removedMap[id] = true;
        });

        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]]) {
                    try {
                        var subflow = this.subflowInstanceNodes[stopList[i]];
                        promises.push(stopNode(node,false).then(() => { subflow.stop() }));
                    } catch(err) {
                        node.error(err);
                    }
                    delete this.subflowInstanceNodes[stopList[i]];
                } else {
                    try {
                        var removed = removedMap[stopList[i]];
                        promises.push(stopNode(node,removed).catch(()=>{}));
                    } catch(err) {
                        node.error(err);
                    }
                }
            }
        }
        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 is a node owned by this flow, so return whatever we have got
            // During a stop/restart, activeNodes could be null for this id
            return this.activeNodes[id];
        } else if (this.activeNodes[id]) {
            // TEMP: this is a subflow internal node within this flow
            return this.activeNodes[id];
        } 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 all of the nodes instantiated within this flow
     * @return {[type]} [description]
     */
    getActiveNodes() {
        return this.activeNodes;
    }

    /**
     * Get a flow setting value. This currently automatically defers to the parent
     * flow which, as defined in ./index.js returns `process.env[key]`.
     * This lays the groundwork for Subflow to have instance-specific settings
     * @param  {[type]} key [description]
     * @return {[type]}     [description]
     */
    getSetting(key) {
        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) {
            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)) {
                    node.users[userNode]._flow.handleStatus(node,statusMessage,node.users[userNode],true);
                }
            }
            handled = true;
        } else {
            this.statusNodes.forEach(function(targetStatusNode) {
                if (targetStatusNode.scope && targetStatusNode.scope.indexOf(reportingNode.id) === -1) {
                    return;
                }
                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 !== null) {
            if (msg.error.hasOwnProperty("source") && msg.error.source !== null) {
                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)) {
                    node.users[userNode]._flow.handleError(node,logMessage,msg,node.users[userNode]);
                }
            }
            handled = true;
        } else {
            var handledByUncaught = false;

            this.catchNodes.forEach(function(targetCatchNode) {
                if (targetCatchNode.scope && targetCatchNode.scope.indexOf(reportingNode.id) === -1) {
                    return;
                }
                if (!targetCatchNode.scope && targetCatchNode.uncaught && !handledByUncaught) {
                    if (handled) {
                        // This has been handled by a !uncaught catch node
                        return;
                    }
                    // This is an uncaught error
                    handledByUncaught = true;
                }
                var errorMessage;
                if (msg) {
                    errorMessage = redUtil.cloneMessage(msg);
                } else {
                    errorMessage = {};
                }
                if (errorMessage.hasOwnProperty("error")) {
                    errorMessage._error = errorMessage.error;
                }
                errorMessage.error = {
                    message: logMessage.toString(),
                    source: {
                        id: node.id,
                        type: node.type,
                        name: node.name,
                        count: count
                    }
                };
                if (logMessage.hasOwnProperty('stack')) {
                    errorMessage.error.stack = logMessage.stack;
                }
                targetCatchNode.receive(errorMessage);
                handled = true;
            });
        }
        return handled;
    }

    dump() {
        console.log("==================")
        console.log(this.TYPE, this.id);
        for (var id in this.activeNodes) {
            if (this.activeNodes.hasOwnProperty(id)) {
                var node = this.activeNodes[id];
                console.log(" ",id.padEnd(16),node.type)
                if (node.wires) {
                    console.log("   -> ",node.wires)
                }
            }
        }
        console.log("==================")
    }

}

/**
 * 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);
    const closeTimeout = new Promise((resolve,reject) => {
        setTimeout(() => {
            reject("Close timed out");
        }, nodeCloseTimeout);
    });
    return Promise.race([closePromise,closeTimeout]).then(() => {
        var delta = Date.now() - start;
        Log.trace("Stopped node "+node.type+":"+node.id+" ("+delta+"ms)" );
    }).catch(err => {
        node.error(Log._("nodes.flows.stopping-error",{message:err}));
        Log.debug(err.stack);
    })
}


module.exports = {
    init: function(runtime) {
        nodeCloseTimeout = runtime.settings.nodeCloseTimeout || 15000;
        Log = runtime.log;
        Subflow = require("./Subflow");
        Subflow.init(runtime);
    },
    create: function(parent,global,conf) {
        return new Flow(parent,global,conf);
    },
    Flow: Flow
}