Merge pull request #2227 from node-red/node-done

Adds Done callback to Input event handler

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

@@ -18,15 +18,27 @@ var util = require("util");
 var EventEmitter = require("events").EventEmitter;
 
 var redUtil = require("@node-red/util").util;
-var Log = require("@node-red/util").log; // TODO: separate module
+var Log = require("@node-red/util").log;
 var context = require("./context");
 var flows = require("./flows");
 
+const NOOP_SEND = function() {}
+
+/**
+ * The Node object is the heart of a Node-RED flow. It is the object that all
+ * nodes extend.
+ *
+ * The Node object itself inherits from EventEmitter, although it provides
+ * custom implementations of some of the EE functions in order to handle
+ * `input` and `close` events properly.
+ */
 function Node(n) {
     this.id = n.id;
     this.type = n.type;
     this.z = n.z;
     this._closeCallbacks = [];
+    this._inputCallback = null;
+    this._inputCallbacks = null;
 
     if (n.name) {
         this.name = n.name;
@@ -43,12 +55,31 @@ function Node(n) {
         // as part of its constructure - config._flow will overwrite this._flow
         // which we can tolerate as they are the same object.
         Object.defineProperty(this,'_flow', {value: n._flow, enumerable: false, writable: true })
+        this._asyncDelivery = n._flow.asyncMessageDelivery;
+    }
+    if (this._asyncDelivery === undefined) {
+        this._asyncDelivery = true;
     }
     this.updateWires(n.wires);
 }
 
 util.inherits(Node, EventEmitter);
 
+/**
+ * Update the wiring configuration for this node.
+ *
+ * We try to optimise the message handling path. To do this there are three
+ * cases to consider:
+ *  1. this node is wired to nothing. In this case we replace node.send with a
+ *     NO-OP function.
+ *  2. this node is wired to one other node. In this case we set `this._wire`
+ *     as a reference to the node it is wired to. This means we avoid unnecessary
+ *     iterations over what would otherwise be a 1-element array.
+ *  3. this node is wired to multiple things. The normal node.send processing of
+ *     this.wires applies.
+ *
+ * @param  {array} wires the new wiring configuration
+ */
 Node.prototype.updateWires = function(wires) {
     //console.log("UPDATE",this.id);
     this.wires = wires || [];
@@ -61,7 +92,7 @@ Node.prototype.updateWires = function(wires) {
     this._wireCount = wc;
     if (wc === 0) {
         // With nothing wired to the node, no-op send
-        this.send = function(msg) {}
+        this.send = NOOP_SEND
     } else {
         this.send = Node.prototype.send;
         if (this.wires.length === 1 && this.wires[0].length === 1) {
@@ -72,6 +103,13 @@ Node.prototype.updateWires = function(wires) {
     }
 
 }
+/**
+ * Get the context object for this node.
+ *
+ * As most nodes do not use context, this is a lazy function that will only
+ * create a context instance for the node if it is needed.
+ * @return {object} the context object
+ */
 Node.prototype.context = function() {
     if (!this._context) {
         this._context = context.get(this._alias||this.id,this.z);
@@ -79,29 +117,194 @@ Node.prototype.context = function() {
     return this._context;
 }
 
+/**
+ * Handle the complete event for a message
+ *
+ * @param  {object} msg  The message that has completed
+ * @param  {error} error (optional) an error hit whilst handling the message
+ */
+Node.prototype._complete = function(msg,error) {
+    if (error) {
+        // For now, delegate this to this.error
+        // But at some point, the timeout handling will need to know about
+        // this as well.
+        this.error(error,msg);
+    } else {
+        this._flow.handleComplete(this,msg);
+    }
+}
+
+/**
+ * An internal reference to the original EventEmitter.on() function
+ */
 Node.prototype._on = Node.prototype.on;
 
+/**
+ * Register a callback function for a named event.
+ * 'close' and 'input' events are handled locally, other events defer to EventEmitter.on()
+ */
 Node.prototype.on = function(event, callback) {
     var node = this;
     if (event == "close") {
         this._closeCallbacks.push(callback);
+    } else if (event === "input") {
+        if (this._inputCallback) {
+            this._inputCallbacks = [this._inputCallback, callback];
+            this._inputCallback = null;
+        } else if (this._inputCallbacks) {
+            this._inputCallbacks.push(callback);
+        } else {
+            this._inputCallback = callback;
+        }
     } else {
         this._on(event, callback);
     }
 };
 
+/**
+ * An internal reference to the original EventEmitter.emit() function
+ */
+Node.prototype._emit = Node.prototype.emit;
+
+/**
+ * Emit an event to all registered listeners.
+ */
+Node.prototype.emit = function(event,arg) {
+    var node = this;
+    if (event === "input") {
+        // When Pluggable Message Routing arrives, this will be called from
+        // that and will already be sync/async depending on the router.
+        if (this._asyncDelivery) {
+            setImmediate(function() {
+                node._emitInput(arg);
+            });
+        } else {
+            this._emitInput(arg);
+        }
+    } else {
+        this._emit(event,arg);
+    }
+}
+
+/**
+ * Handle the 'input' event.
+ *
+ * This will call all registered handlers for the 'input' event.
+ */
+Node.prototype._emitInput = function(arg) {
+    var node = this;
+    if (node._inputCallback) {
+        // Just one callback registered.
+        try {
+            node._inputCallback(
+                arg,
+                function() { node.send.apply(node,arguments) },
+                function(err) { node._complete(arg,err); }
+            );
+        } catch(err) {
+            node.error(err,arg);
+        }
+    } else if (node._inputCallbacks) {
+        // Multiple callbacks registered. Call each one, tracking eventual completion
+        var c = node._inputCallbacks.length;
+        for (var i=0;i<c;i++) {
+            var cb = node._inputCallbacks[i];
+            if (cb.length === 2) {
+                c++;
+            }
+            try {
+                node._inputCallbacks[i](
+                    arg,
+                    function() { node.send.apply(node,arguments) },
+                    function(err) {
+                        c--;
+                        if (c === 0) {
+                            node._complete(arg,err);
+                        }
+                    }
+                );
+            } catch(err) {
+                node.error(err,msg);
+            }
+        }
+    }
+}
+
+/**
+ * An internal reference to the original EventEmitter.removeListener() function
+ */
+Node.prototype._removeListener = Node.prototype.removeListener;
+
+/**
+ * Remove a listener for an event
+ */
+Node.prototype.removeListener = function(name, listener) {
+    var index;
+    if (name === "input") {
+        if (this._inputCallback && this._inputCallback === listener) {
+            // Removing the only callback
+            this._inputCallback = null;
+        } else if (this._inputCallbacks) {
+            // Removing one of many callbacks
+            index = this._inputCallbacks.indexOf(listener);
+            if (index > -1) {
+                this._inputCallbacks.splice(index,1);
+            }
+            // Check if we can optimise back to a single callback
+            if (this._inputCallbacks.length === 1) {
+                this._inputCallback = this._inputCallbacks[0];
+                this._inputCallbacks = null;
+            }
+        }
+    } else if (name === "close") {
+        index = this._closeCallbacks.indexOf(listener);
+        if (index > -1) {
+            this._closeCallbacks.splice(index,1);
+        }
+    } else {
+        this._removeListener(name, listener);
+    }
+}
+
+/**
+ * An internal reference to the original EventEmitter.removeAllListeners() function
+ */
+Node.prototype._removeAllListeners = Node.prototype.removeAllListeners;
+
+/**
+ * Remove all listeners for an event
+ */
+Node.prototype.removeAllListeners = function(name) {
+    if (name === "input") {
+        this._inputCallback = null;
+        this._inputCallbacks = null;
+    } else if (name === "close") {
+        this._closeCallbacks = [];
+    } else {
+        this._removeAllListeners(name);
+    }
+}
+
+/**
+ * Called when the node is being stopped
+ * @param  {boolean} removed Whether the node has been removed, or just being stopped
+ * @return {Promise} resolves when the node has closed
+ */
 Node.prototype.close = function(removed) {
     //console.log(this.type,this.id,removed);
     var promises = [];
     var node = this;
+    // Call all registered close callbacks.
     for (var i=0;i<this._closeCallbacks.length;i++) {
         var callback = this._closeCallbacks[i];
         if (callback.length > 0) {
+            // The callback takes a 'done' callback and (maybe) the removed flag
             promises.push(
                 new Promise((resolve) => {
                     try {
                         var args = [];
                         if (callback.length === 2) {
+                            // The listener expects the removed flag
                             args.push(!!removed);
                         }
                         args.push(() => {
@@ -116,6 +319,7 @@ Node.prototype.close = function(removed) {
                 })
             );
         } else {
+            // No done callback so handle synchronously
             try {
                 callback.call(node);
             } catch(err) {
@@ -138,6 +342,12 @@ Node.prototype.close = function(removed) {
     }
 };
 
+/**
+ * Send a message to the nodes wired.
+ *
+ *
+ * @param  {object} msg A message or array of messages to send
+ */
 Node.prototype.send = function(msg) {
     var msgSent = false;
     var node;
@@ -225,6 +435,12 @@ Node.prototype.send = function(msg) {
     }
 };
 
+/**
+ * Receive a message.
+ *
+ * This will emit the `input` event with the provided message.
+ * As of 1.0, this will return *before* any 'input' callback handler is invoked.
+ */
 Node.prototype.receive = function(msg) {
     if (!msg) {
         msg = {};
@@ -233,11 +449,7 @@ Node.prototype.receive = function(msg) {
         msg._msgid = redUtil.generateId();
     }
     this.metric("receive",msg);
-    try {
-        this.emit("input", msg);
-    } catch(err) {
-        this.error(err,msg);
-    }
+    this.emit("input",msg);
 };
 
 function log_helper(self, level, msg) {
@@ -258,15 +470,23 @@ function log_helper(self, level, msg) {
     }
     Log.log(o);
 }
-
+/**
+ * Log an INFO level message
+ */
 Node.prototype.log = function(msg) {
     log_helper(this, Log.INFO, msg);
 };
 
+/**
+ * Log a WARN level message
+ */
 Node.prototype.warn = function(msg) {
     log_helper(this, Log.WARN, msg);
 };
 
+/**
+ * Log an ERROR level message
+ */
 Node.prototype.error = function(logMessage,msg) {
     if (typeof logMessage != 'boolean') {
         logMessage = logMessage || "";
@@ -280,15 +500,22 @@ Node.prototype.error = function(logMessage,msg) {
     }
 };
 
+/**
+ * Log an DEBUG level message
+ */
 Node.prototype.debug = function(msg) {
     log_helper(this, Log.DEBUG, msg);
 }
 
+/**
+ * Log an TRACE level message
+ */
 Node.prototype.trace = function(msg) {
     log_helper(this, Log.TRACE, msg);
 }
 
 /**
+ * Log a metric event.
  * If called with no args, returns whether metric collection is enabled
  */
 Node.prototype.metric = function(eventname, msg, metricValue) {
@@ -305,6 +532,8 @@ Node.prototype.metric = function(eventname, msg, metricValue) {
 }
 
 /**
+ * Set the node's status object
+ *
  * status: { fill:"red|green", shape:"dot|ring", text:"blah" }
  * or
  * status: "simple text status"

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

@@ -23,6 +23,7 @@ var Subflow;
 var Log;
 
 var nodeCloseTimeout = 15000;
+var asyncMessageDelivery = true;
 
 /**
  * This class represents a flow within the runtime. It is responsible for
@@ -125,6 +126,7 @@ class Flow {
         var id;
         this.catchNodes = [];
         this.statusNodes = [];
+        this.completeNodeMap = {};
 
         var configNodes = Object.keys(this.flow.configs);
         var configNodeAttempts = {};
@@ -228,7 +230,7 @@ class Flow {
             this.trace(" id               | type         | alias");
             this.trace("------------------|--------------|-----------------");
         }
-        // Build the map of catch/status nodes.
+        // Build the map of catch/status/complete nodes.
         for (id in this.activeNodes) {
             if (this.activeNodes.hasOwnProperty(id)) {
                 node = this.activeNodes[id];
@@ -237,6 +239,13 @@ class Flow {
                     this.catchNodes.push(node);
                 } else if (node.type === "status") {
                     this.statusNodes.push(node);
+                } else if (node.type === "complete") {
+                    if (node.scope) {
+                        node.scope.forEach(id => {
+                            this.completeNodeMap[id] = this.completeNodeMap[id] || [];
+                            this.completeNodeMap[id].push(node);
+                        })
+                    }
                 }
             }
         }
@@ -516,6 +525,20 @@ class Flow {
         return handled;
     }
 
+    handleComplete(node,msg) {
+        if (this.completeNodeMap[node.id]) {
+            let toSend = msg;
+            this.completeNodeMap[node.id].forEach((completeNode,index) => {
+                toSend = redUtil.cloneMessage(msg);
+                completeNode.receive(toSend);
+            })
+        }
+    }
+
+    get asyncMessageDelivery() {
+        return asyncMessageDelivery
+    }
+
     dump() {
         console.log("==================")
         console.log(this.TYPE, this.id);
@@ -562,6 +585,7 @@ function stopNode(node,removed) {
 module.exports = {
     init: function(runtime) {
         nodeCloseTimeout = runtime.settings.nodeCloseTimeout || 15000;
+        asyncMessageDelivery = !runtime.settings.runtimeSyncDelivery
         Log = runtime.log;
         Subflow = require("./Subflow");
         Subflow.init(runtime);

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

@@ -729,6 +729,10 @@ module.exports = {
     updateFlow: updateFlow,
     removeFlow: removeFlow,
     disableFlow:null,
-    enableFlow:null
+    enableFlow:null,
+    isDeliveryModeAsync: function() {
+        // If settings is null, this is likely being run by unit tests
+        return !settings || !settings.runtimeSyncDelivery
+    }
 
 };