node-red/packages/node_modules/@node-red/nodes/core/common/60-link.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.
 **/

module.exports = function(RED) {
    "use strict";
    const crypto = require("crypto");
    const targetCache = (function() {
        const name2id = {};
        let id2target = {};
        return {
            /** @type {((node: Node, [flowName]: string) => string) & ((nodeName: string, flowName: string) => string))} */
            generateLookupName(node, flowName) {
                if(!flowName) {
                    flowName = node._flow.flow.label
                }
                if(node instanceof LinkInNode) {
                    return `${flowName}/${node.name}`;
                }
                return `${flowName}/${node}`;
            },
            add(node) {
                const id = node.id;
                const nodeName = node.name;
                if(!nodeName){ return null;} //node must be named
                const flowName = node._flow.flow.label;
                const lookupName = targetCache.generateLookupName(nodeName, flowName);
                console.log(`Adding node '${lookupName}' (${id}) to cache`)
                if(name2id[lookupName] && name2id[lookupName].id !== id) {
                    //TODO: reassignment? duplate lookupName?  throw error or warning?
                    console.warn(`💣 '${lookupName}' is already assigned to node with ID: ${name2id[lookupName]} `)
                }
                
                name2id[lookupName] = id;
                id2target[id] = {
                    lookupName,
                    nodeName,
                    flowName,
                    id
                }
                return id2target[id];
            },
            remove(node) {
                if(node.name) console.log(`Removing node '${targetCache.generateLookupName(node)}' (${node.id})`)
                const target = id2target[node.id];
                targetCache._removeById(node.id);
                if(target) {
                    targetCache._removeById(target.id);
                    targetCache._removeByName(target.lookupName);
                }
            },
            getTarget(lookupName) {
                const id = name2id[lookupName];
                const target = id2target[id];
                return target;
            },
            verify(cachedTarget, targetNode, nodeName, flowName) {
                cachedTarget = cachedTarget || {};
                targetNode = targetNode || {};
                const _lookupName = targetCache.generateLookupName(nodeName, flowName);
                const cachedId = name2id[_lookupName];
   
                const idOK = cachedTarget.id === targetNode.id && cachedId === targetNode.id;
                const nodeNameOK = nodeName === cachedTarget.nodeName && cachedTarget.nodeName == targetNode.name;
                const flowNameOK = flowName === cachedTarget.flowName && cachedTarget.flowName == targetNode._flow.flow.label;
                const lookupNameOK = _lookupName === cachedTarget.lookupName;
                if(idOK && nodeNameOK && flowNameOK && lookupNameOK) {
                    return true;
                }
                console.warn(`verify(cachedTarget, targetNode, nodeName, flowName) failed: `, {cachedTarget, targetNode, nodeName, flowName})
                console.warn(`> idOK:${idOK}, nodeNameOK:${nodeNameOK}, flowNameOK:${flowNameOK}, lookupNameOK:${lookupNameOK}`)
                targetCache._removeById(targetNode.id);
                targetCache._removeById(cachedId);
                targetCache._removeByName(_lookupName);
                targetCache._removeByName(cachedTarget.lookupName);
                return false;
            },
            _removeById(id) {
                if(!id) {
                    return;
                }
                const target = id2target[id];
                if(target && target.lookupName) {
                    delete name2id[target.lookupName];
                }
                delete id2target[id];
            },
            _removeByName(lookupName) {
                if(!lookupName) {
                    return;
                }
                var id = name2id[lookupName];
                var target = id2target[id];
                if(id) {
                    delete id2target[id];
                }
                if(target && target.lookupName) {
                    delete name2id[target.lookupName];
                }
                delete name2id[lookupName];
            },
            clear() {
                name2id = {};
                id2target = {};
            }
        }
    })();

    function LinkInNode(n) {
        RED.nodes.createNode(this,n);
        var node = this;
        var event = "node:"+n.id;
        var handler = function(msg) {
            msg._event = n.event;
            node.receive(msg);
        }
        //update target cache for link calls
        function updateCache(o) {
            console.log(node.id, node.name, o.config.rev, o)
            const changed = o.diff.changed.includes(node.id);
            const added = o.diff.added.includes(node.id);
            if(changed) {
                targetCache.remove(node);
            }
            if((changed || added) && node.name) {
                targetCache.add(node);
            }
        }

        RED.events.on("flows:started", updateCache);
        RED.events.on(event,handler);
        this.on("input", function(msg, send, done) {
            send(msg);
            done();
        });
        this.on("close",function() {
            targetCache.remove(node);
            RED.events.removeListener(event,handler);
        });
    }

    RED.nodes.registerType("link in",LinkInNode);

    function LinkOutNode(n) {
        RED.nodes.createNode(this,n);
        var node = this;
        var mode = n.mode || "link";

        var event = "node:"+n.id;
        this.on("input", function(msg, send, done) {
            msg._event = event;
            RED.events.emit(event,msg)

            if (mode === "return") {
                if (Array.isArray(msg._linkSource) && msg._linkSource.length > 0) {
                    var messageEvent = msg._linkSource.pop();
                    var returnNode = RED.nodes.getNode(messageEvent.node);
                    if (returnNode && returnNode.returnLinkMessage) {
                        returnNode.returnLinkMessage(messageEvent.id, msg);
                    } else {
                        node.warn(RED._("link.error.missingReturn"))
                    }
                } else {
                    node.warn(RED._("link.error.missingReturn"))
                }
                done();
            } else if (mode === "link") {
                send(msg);
                done();
            }
        });
    }
    RED.nodes.registerType("link out",LinkOutNode);


    function LinkCallNode(n) {
        RED.nodes.createNode(this,n);
        const node = this;
        const staticTarget = n.links[0];
        const linkType = n.linkType;
        const messageEvents = {};

        let timeout = parseFloat(n.timeout || "30") * 1000;
        if (isNaN(timeout)) {
            timeout = 30000;
        }
        function findNode(target) {
            let foundNode = null;
            switch (typeof target) {
                case "string":
                    const nameParts = target.split("/");
                    switch (nameParts.length) {
                        case 1:
                            target = { nodeName: nameParts[0], flowName: node._flow.flow.label }
                            break;
                        case 2:
                            target = { nodeName: nameParts[1], flowName: nameParts[0] }
                            break;
                        default:
                            target = { nodeName: target, flowName: node._flow.flow.label }
                    }
                case "object":
                    break;
                default:
                    throw new Error("Invalid target");
            }
            if(!target.nodeName || typeof target.nodeName !== "string") {
                throw new Error("Link call target name invalid")
            }
            if(!target.flowName || typeof target.flowName !== "string") {
                throw new Error("Link call target flow name invalid")
            }
            const lookupName = targetCache.generateLookupName(target.nodeName, target.flowName);
            //const targetNode = targetCache.getNode(lookupName);
            const cachedTarget = targetCache.getTarget(lookupName);
            if (cachedTarget) {
                foundNode = RED.nodes.getNode(cachedTarget.id);
                if (targetCache.verify(cachedTarget, foundNode, target.nodeName, target.flowName)) {
                    return foundNode;
                }
            }
            throw new Error(`Link in node not found: ${lookupName}`);
        }
        this.on("input", function(msg, send, done) {
            try {
                let targetNode = linkType == "dynamic" ? findNode(msg.target) : RED.nodes.getNode(staticTarget);
                if (targetNode && targetNode instanceof LinkInNode) {
                    msg._linkSource = msg._linkSource || [];
                    const messageEvent = {
                        id: crypto.randomBytes(14).toString('hex'),
                        node: node.id,
                    }        
                    messageEvents[messageEvent.id] = {
                        msg: RED.util.cloneMessage(msg),
                        send,
                        done,
                        ts: setTimeout(function() {
                            timeoutMessage(messageEvent.id)
                        }, timeout )
                    };
                    msg._linkSource.push(messageEvent);
                    targetNode.receive(msg);
                }  
            } catch (error) {
                node.error(error, msg);
            }
        });

        this.returnLinkMessage = function(eventId, msg) {
            if (Array.isArray(msg._linkSource) && msg._linkSource.length === 0) {
                delete msg._linkSource;
            }
            const messageEvent = messageEvents[eventId];
            if (messageEvent) {
                clearTimeout(messageEvent.ts);
                delete messageEvents[eventId];
                messageEvent.send(msg);
                messageEvent.done();
            } else {
                node.send(msg);
            }
        }

        function timeoutMessage(eventId) {
            const messageEvent = messageEvents[eventId];
            if (messageEvent) {
                delete messageEvents[eventId];
                node.error("timeout",messageEvent.msg);
            }
        }

    }
    RED.nodes.registerType("link call",LinkCallNode);


}