Merge branch 'async-split' into 0.19

nodes/core/logic/17-split.js

@@ -233,7 +233,7 @@ module.exports = function(RED) {
     RED.nodes.registerType("split",SplitNode);
 
 
-    var _max_kept_msgs_count = undefined;
+    var _max_kept_msgs_count;
 
     function max_kept_msgs_count(node) {
         if (_max_kept_msgs_count === undefined) {
@@ -252,7 +252,15 @@ module.exports = function(RED) {
         exp.assign("I", index);
         exp.assign("N", count);
         exp.assign("A", accum);
-        return RED.util.evaluateJSONataExpression(exp, msg);
+        return new Promise((resolve,reject) => {
+            RED.util.evaluateJSONataExpression(exp, msg, (err, result) => {
+                if (err) {
+                    reject(err);
+                } else {
+                    resolve(result);
+                }
+            });
+        });
     }
 
     function apply_f(exp, accum, count) {
@@ -269,11 +277,11 @@ module.exports = function(RED) {
         return exp
     }
 
-    function reduce_and_send_group(node, group) {
+    function reduceAndSendGroup(node, group) {
         var is_right = node.reduce_right;
         var flag = is_right ? -1 : 1;
         var msgs = group.msgs;
-        var accum = eval_exp(node, node.exp_init, node.exp_init_type);
+        return getInitialReduceValue(node, node.exp_init, node.exp_init_type).then(accum => {
             var reduce_exp = node.reduce_exp;
             var reduce_fixup = node.reduce_fixup;
             var count = group.count;
@@ -284,16 +292,21 @@ module.exports = function(RED) {
                 if (ix > iy) {return flag;}
                 return 0;
             });
-        for(var msg of msgs) {
+
-            accum = apply_r(reduce_exp, accum, msg, msg.parts.index, count);
+            return msgs.reduce((promise, msg) => promise.then(accum => apply_r(reduce_exp, accum, msg, msg.parts.index, count)), Promise.resolve(accum))
-        }
+                .then(accum => {
                     if(reduce_fixup !== undefined) {
                         accum = apply_f(reduce_fixup, accum, count);
                     }
                     node.send({payload: accum});
+                });
+        }).catch(err => {
+            throw new Error(RED._("join.errors.invalid-expr",{error:e.message}));
+        });
     }
 
     function reduce_msg(node, msg) {
+        var promise;
         if (msg.hasOwnProperty('parts')) {
             var parts = msg.parts;
             var pending = node.pending;
@@ -317,60 +330,77 @@ module.exports = function(RED) {
             }
             msgs.push(msg);
             pending_count++;
-            if(msgs.length === group.count) {
+            var completeProcess = function() {
-                delete pending[gid];
-                try {
-                    pending_count -= msgs.length;
-                    reduce_and_send_group(node, group);
-                } catch(e) {
-                    node.error(RED._("join.errors.invalid-expr",{error:e.message}));            }
-            }
                 node.pending_count = pending_count;
                 var max_msgs = max_kept_msgs_count(node);
                 if ((max_msgs > 0) && (pending_count > max_msgs)) {
                     node.pending = {};
                     node.pending_count = 0;
-                node.error(RED._("join.too-many"), msg);
+                    var promise = Promise.reject(RED._("join.too-many"));
+                    promise.catch(()=>{});
+                    return promise;
                 }
+                return Promise.resolve();
             }
-        else {
+            if(msgs.length === group.count) {
+                delete pending[gid];
+                pending_count -= msgs.length;
+                promise = reduceAndSendGroup(node, group).then(completeProcess);
+            } else {
+                promise = completeProcess();
+            }
+        } else {
             node.send(msg);
         }
+        if (!promise) {
+            promise = Promise.resolve();
+        }
+        return promise;
     }
 
-    function eval_exp(node, exp, exp_type) {
+    function getInitialReduceValue(node, exp, exp_type) {
-        if(exp_type === "flow") {
+        return new Promise((resolve,reject) => {
-            return node.context().flow.get(exp);
+            if(exp_type === "flow" || exp_type === "global") {
+                node.context()[exp_type].get(exp,(err,value) => {
+                    if (err) {
+                        reject(err);
+                    } else {
+                        resolve(value);
                     }
-        else if(exp_type === "global") {
+                });
-            return node.context().global.get(exp);
+                return;
-        }
+            } else if(exp_type === "jsonata") {
-        else if(exp_type === "str") {
-            return exp;
-        }
-        else if(exp_type === "num") {
-            return Number(exp);
-        }
-        else if(exp_type === "bool") {
-            if (exp === 'true') {
-                return true;
-            }
-            else if (exp === 'false') {
-                return false;
-            }
-        }
-        else if ((exp_type === "bin") ||
-                 (exp_type === "json")) {
-            return JSON.parse(exp);
-        }
-        else if(exp_type === "date") {
-            return Date.now();
-        }
-        else if(exp_type === "jsonata") {
                 var jexp = RED.util.prepareJSONataExpression(exp, node);
-            return RED.util.evaluateJSONataExpression(jexp, {});
+                RED.util.evaluateJSONataExpression(jexp, {},(err,value) => {
+                    if (err) {
+                        reject(err);
+                    } else {
+                        resolve(value);
                     }
-        throw new Error("unexpected initial value type");
+                });
+                return;
+            }
+            var result;
+            if(exp_type === "str") {
+                result = exp;
+            } else if(exp_type === "num") {
+                result = Number(exp);
+            } else if(exp_type === "bool") {
+                if (exp === 'true') {
+                    result = true;
+                } else if (exp === 'false') {
+                    result = false;
+                }
+            } else if ((exp_type === "bin") || (exp_type === "json")) {
+                result = JSON.parse(exp);
+            } else if(exp_type === "date") {
+                result = Date.now();
+            } else {
+                reject(new Error("unexpected initial value type"));
+                return;
+            }
+            resolve(result);
+        });
     }
 
     function JoinNode(n) {
@@ -482,6 +512,40 @@ module.exports = function(RED) {
             node.send(group.msg);
         }
 
+        var pendingMessages = [];
+        var activeMessagePromise = null;
+        // In reduce mode, we must process messages fully in order otherwise
+        // groups may overlap and cause unexpected results. The use of JSONata
+        // means some async processing *might* occur if flow/global context is
+        // accessed.
+        var processReduceMessageQueue = 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 = reduce_msg(node, nextMsg)
+                .then(processReduceMessageQueue)
+                .catch((err) => {
+                    node.error(err,nextMsg);
+                    return processReduceMessageQueue();
+                });
+        }
+
         this.on("input", function(msg) {
             try {
                 var property;
@@ -520,8 +584,7 @@ module.exports = function(RED) {
                     propertyIndex = msg.parts.index;
                 }
                 else if (node.mode === 'reduce') {
-                    reduce_msg(node, msg);
+                    return processReduceMessageQueue(msg);
-                    return;
                 }
                 else {
                     // Use the node configuration to identify all of the group information