node-red/nodes/core/logic/10-switch.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";

    var operators = {
        'eq': function(a, b) { return a == b; },
        'neq': function(a, b) { return a != b; },
        'lt': function(a, b) { return a < b; },
        'lte': function(a, b) { return a <= b; },
        'gt': function(a, b) { return a > b; },
        'gte': function(a, b) { return a >= b; },
        'btwn': function(a, b, c) { return a >= b && a <= c; },
        'cont': function(a, b) { return (a + "").indexOf(b) != -1; },
        'regex': function(a, b, c, d) { return (a + "").match(new RegExp(b,d?'i':'')); },
        'true': function(a) { return a === true; },
        'false': function(a) { return a === false; },
        'null': function(a) { return (typeof a == "undefined" || a === null); },
        'nnull': function(a) { return (typeof a != "undefined" && a !== null); },
        'istype': function(a, b) {
            if (b === "array") { return Array.isArray(a); }
            else if (b === "buffer") { return Buffer.isBuffer(a); }
            else if (b === "json") {
                try { JSON.parse(a); return true; }   // or maybe ??? a !== null; }
                catch(e) { return false;}
            }
            else if (b === "null") { return a === null; }
            else { return typeof a === b && !Array.isArray(a) && !Buffer.isBuffer(a) && a !== null; }
        },
        'head': function(a, b, c, d, parts) {
            var count = Number(b);
            return (parts.index < count);
        },
        'tail': function(a, b, c, d, parts) {
            var count = Number(b);
            return (parts.count -count <= parts.index);
        },
        'index': function(a, b, c, d, parts) {
            var min = Number(b);
            var max = Number(c);
            var index = parts.index;
            return ((min <= index) && (index <= max));
        },
        'jsonata_exp': function(a, b) { return (b === true); },
        'else': function(a) { return a === true; }
    };

    var _maxKeptCount;

    function getMaxKeptCount() {
        if (_maxKeptCount === undefined) {
            var name = "nodeMessageBufferMaxLength";
            if (RED.settings.hasOwnProperty(name)) {
                _maxKeptCount = RED.settings[name];
            }
            else {
                _maxKeptCount = 0;
            }
        }
        return _maxKeptCount;
    }

    function getProperty(node,msg) {
        return new Promise((resolve,reject) => {
            if (node.propertyType === 'jsonata') {
                RED.util.evaluateJSONataExpression(node.property,msg,(err,value) => {
                    if (err) {
                        reject(RED._("switch.errors.invalid-expr",{error:err.message}));
                    } else {
                        resolve(value);
                    }
                });
            } else {
                RED.util.evaluateNodeProperty(node.property,node.propertyType,node,msg,(err,value) => {
                    if (err) {
                        resolve(undefined);
                    } else {
                        resolve(value);
                    }
                });
            }
        });
    }

    function getV1(node,msg,rule,hasParts) {
        return new Promise( (resolve,reject) => {
            if (rule.vt === 'prev') {
                resolve(node.previousValue);
            } else if (rule.vt === 'jsonata') {
                var exp = rule.v;
                if (rule.t === 'jsonata_exp') {
                    if (hasParts) {
                        exp.assign("I", msg.parts.index);
                        exp.assign("N", msg.parts.count);
                    }
                }
                RED.util.evaluateJSONataExpression(exp,msg,(err,value) => {
                    if (err) {
                        reject(RED._("switch.errors.invalid-expr",{error:err.message}));
                    } else {
                        resolve(value);
                    }
                });
            } else if (rule.vt === 'json') {
                resolve("json");
            } else if (rule.vt === 'null') {
                resolve("null");
            } else {
                RED.util.evaluateNodeProperty(rule.v,rule.vt,node,msg, function(err,value) {
                    if (err) {
                        resolve(undefined);
                    } else {
                        resolve(value);
                    }
                });
            }
        });
    }

    function getV2(node,msg,rule) {
        return new Promise((resolve,reject) => {
            var v2 = rule.v2;
            if (rule.v2t === 'prev') {
                resolve(node.previousValue);
            } else if (rule.v2t === 'jsonata') {
                RED.util.evaluateJSONataExpression(rule.v2,msg,(err,value) => {
                    if (err) {
                        reject(RED._("switch.errors.invalid-expr",{error:err.message}));
                    } else {
                        resolve(value);
                    }
                });
            } else if (typeof v2 !== 'undefined') {
                RED.util.evaluateNodeProperty(rule.v2,rule.v2t,node,msg, function(err,value) {
                    if (err) {
                        resolve(undefined);
                    } else {
                        resolve(value);
                    }
                });
            } else {
                resolve(v2);
            }
        })
    }

    function applyRule(node, msg, property, state) {
        return new Promise((resolve,reject) => {

            var rule = node.rules[state.currentRule];
            var v1,v2;

            getV1(node,msg,rule,state.hasParts).then(value => {
                v1 = value;
            }).then(()=>getV2(node,msg,rule)).then(value => {
                v2 = value;
            }).then(() => {
                if (rule.t == "else") {
                    property = state.elseflag;
                    state.elseflag = true;
                }
                if (operators[rule.t](property,v1,v2,rule.case,msg.parts)) {
                    state.onward.push(msg);
                    state.elseflag = false;
                    if (node.checkall == "false") {
                        return resolve(false);
                    }
                } else {
                    state.onward.push(null);
                }
                resolve(state.currentRule < node.rules.length - 1);
            });
        })
    }

    function applyRules(node, msg, property,state) {
        if (!state) {
            state = {
                currentRule: 0,
                elseflag: true,
                onward: [],
                hasParts: msg.hasOwnProperty("parts") &&
                                msg.parts.hasOwnProperty("id") &&
                                msg.parts.hasOwnProperty("index")
            }
        }
        return applyRule(node,msg,property,state).then(hasMore => {
            if (hasMore) {
                state.currentRule++;
                return applyRules(node,msg,property,state);
            } else {
                node.previousValue = property;
                return state.onward;
            }
        });
    }


    function SwitchNode(n) {
        RED.nodes.createNode(this, n);
        this.rules = n.rules || [];
        this.property = n.property;
        this.propertyType = n.propertyType || "msg";

        if (this.propertyType === 'jsonata') {
            try {
                this.property = RED.util.prepareJSONataExpression(this.property,this);
            } catch(err) {
                this.error(RED._("switch.errors.invalid-expr",{error:err.message}));
                return;
            }
        }

        this.checkall = n.checkall || "true";
        this.previousValue = null;
        var node = this;
        var valid = true;
        var repair = n.repair;
        var needsCount = repair;
        for (var i=0; i<this.rules.length; i+=1) {
            var rule = this.rules[i];
            needsCount = needsCount || ((rule.t === "tail") || (rule.t === "jsonata_exp"));
            if (!rule.vt) {
                if (!isNaN(Number(rule.v))) {
                    rule.vt = 'num';
                } else {
                    rule.vt = 'str';
                }
            }
            if (rule.vt === 'num') {
                if (!isNaN(Number(rule.v))) {
                    rule.v = Number(rule.v);
                }
            } else if (rule.vt === "jsonata") {
                try {
                    rule.v = RED.util.prepareJSONataExpression(rule.v,node);
                } catch(err) {
                    this.error(RED._("switch.errors.invalid-expr",{error:err.message}));
                    valid = false;
                }
            }
            if (typeof rule.v2 !== 'undefined') {
                if (!rule.v2t) {
                    if (!isNaN(Number(rule.v2))) {
                        rule.v2t = 'num';
                    } else {
                        rule.v2t = 'str';
                    }
                }
                if (rule.v2t === 'num') {
                    rule.v2 = Number(rule.v2);
                } else if (rule.v2t === 'jsonata') {
                    try {
                        rule.v2 = RED.util.prepareJSONataExpression(rule.v2,node);
                    } catch(err) {
                        this.error(RED._("switch.errors.invalid-expr",{error:err.message}));
                        valid = false;
                    }
                }
            }
        }

        if (!valid) {
            return;
        }

        var pendingCount = 0;
        var pendingId = 0;
        var pendingIn = {};
        var pendingOut = {};
        var received = {};

        function addMessageToGroup(id, msg, parts) {
            if (!(id in pendingIn)) {
                pendingIn[id] = {
                    count: undefined,
                    msgs: [],
                    seq_no: pendingId++
                };
            }
            var group = pendingIn[id];
            group.msgs.push(msg);
            pendingCount++;
            var max_msgs = getMaxKeptCount();
            if ((max_msgs > 0) && (pendingCount > max_msgs)) {
                clearPending();
                node.error(RED._("switch.errors.too-many"), msg);
            }
            if (parts.hasOwnProperty("count")) {
                group.count = parts.count;
            }
            return group;
        }


        function addMessageToPending(msg) {
            var parts = msg.parts;
            // We've already checked the msg.parts has the require bits
            var group = addMessageToGroup(parts.id, msg, parts);
            var msgs = group.msgs;
            var count = group.count;
            if (count === msgs.length) {
                // We have a complete group - send the individual parts
                return msgs.reduce((promise, msg) => {
                    return promise.then((result) => {
                        msg.parts.count = count;
                        return processMessage(msg, false);
                    })
                }, Promise.resolve()).then( () => {
                    pendingCount -= group.msgs.length;
                    delete pendingIn[parts.id];
                });
            }
            return Promise.resolve();
        }

        function sendGroup(onwards, port_count) {
            var counts = new Array(port_count).fill(0);
            for (var i = 0; i < onwards.length; i++) {
                var onward = onwards[i];
                for (var j = 0; j < port_count; j++) {
                    counts[j] += (onward[j] !== null) ? 1 : 0
                }
            }
            var ids = new Array(port_count);
            for (var j = 0; j < port_count; j++) {
                ids[j] = RED.util.generateId();
            }
            var ports = new Array(port_count);
            var indexes = new Array(port_count).fill(0);
            for (var i = 0; i < onwards.length; i++) {
                var onward = onwards[i];
                for (var j = 0; j < port_count; j++) {
                    var msg = onward[j];
                    if (msg) {
                        var new_msg = RED.util.cloneMessage(msg);
                        var parts = new_msg.parts;
                        parts.id = ids[j];
                        parts.index = indexes[j];
                        parts.count = counts[j];
                        ports[j] = new_msg;
                        indexes[j]++;
                    }
                    else {
                        ports[j] = null;
                    }
                }
                node.send(ports);
            }
        }

        function sendGroupMessages(onward, msg) {
            var parts = msg.parts;
            var gid = parts.id;
            received[gid] = ((gid in received) ? received[gid] : 0) +1;
            var send_ok = (received[gid] === parts.count);

            if (!(gid in pendingOut)) {
                pendingOut[gid] = {
                    onwards: []
                };
            }
            var group = pendingOut[gid];
            var onwards = group.onwards;
            onwards.push(onward);
            pendingCount++;
            if (send_ok) {
                sendGroup(onwards, onward.length, msg);
                pendingCount -= onward.length;
                delete pendingOut[gid];
                delete received[gid];
            }
            var max_msgs = getMaxKeptCount();
            if ((max_msgs > 0) && (pendingCount > max_msgs)) {
                clearPending();
                node.error(RED._("switch.errors.too-many"), msg);
            }
        }





        function processMessage(msg, checkParts) {
            var hasParts = msg.hasOwnProperty("parts") &&
                            msg.parts.hasOwnProperty("id") &&
                            msg.parts.hasOwnProperty("index");

            if (needsCount && checkParts && hasParts) {
                return addMessageToPending(msg);
            }
            return getProperty(node,msg)
                    .then(property => applyRules(node,msg,property))
                    .then(onward => {
                        if (!repair || !hasParts) {
                            node.send(onward);
                        }
                        else {
                            sendGroupMessages(onward, msg);
                        }
                    }).catch(err => {
                        node.warn(err);
                    });
        }

        function clearPending() {
            pendingCount = 0;
            pendingId = 0;
            pendingIn = {};
            pendingOut = {};
            received = {};
        }

        var pendingMessages = [];
        var activeMessagePromise = null;
        var processMessageQueue = function(msg) {
            if (msg) {
                // A new message has arrived - add it to the message queue
                pendingMessages.push(msg);
                if (activeMessagePromise !== null) {
                    // The node is currently processing a message, so do nothing
                    // more with this message
                    return;
                }
            }
            if (pendingMessages.length === 0) {
                // There are no more messages to process, clear the active flag
                // and return
                activeMessagePromise = null;
                return;
            }

            // There are more messages to process. Get the next message and
            // start processing it. Recurse back in to check for any more
            var nextMsg = pendingMessages.shift();
            activeMessagePromise = processMessage(nextMsg,true)
                .then(processMessageQueue)
                .catch((err) => {
                    node.error(err,nextMsg);
                    return processMessageQueue();
                });
        }

        this.on('input', function(msg) {
            processMessageQueue(msg);
        });

        this.on('close', function() {
            clearPending();
        });
    }

    RED.nodes.registerType("switch", SwitchNode);
}