Update Function node to add node.done

packages/node_modules/@node-red/nodes/core/core/80-function.js

@@ -19,7 +19,7 @@ module.exports = function(RED) {
     var util = require("util");
     var vm = require("vm");
 
-    function sendResults(node,_msgid,msgs) {
+    function sendResults(node,send,_msgid,msgs) {
         if (msgs == null) {
             return;
         } else if (!util.isArray(msgs)) {
@@ -49,7 +49,7 @@ module.exports = function(RED) {
             }
         }
         if (msgCount>0) {
-            node.send(msgs);
+            send(msgs);
         }
     }
 
@@ -58,8 +58,17 @@ module.exports = function(RED) {
         var node = this;
         this.name = n.name;
         this.func = n.func;
+
+        var handleNodeDoneCall = true;
+        // Check to see if the Function appears to call `node.done()`. If so,
+        // we will assume it is well written and does actually call node.done().
+        // Otherwise, we will call node.done() after the function returns regardless.
+        if (/node\.done\s*\(\s*\)/.test(this.func)) {
+            handleNodeDoneCall = false;
+        }
+
         var functionText = "var results = null;"+
-                           "results = (function(msg){ "+
+                           "results = (function(msg,__send__,__done__){ "+
                               "var __msgid__ = msg._msgid;"+
                               "var node = {"+
                                  "id:__node__.id,"+
@@ -71,10 +80,11 @@ module.exports = function(RED) {
                                  "trace:__node__.trace,"+
                                  "on:__node__.on,"+
                                  "status:__node__.status,"+
-                                 "send:function(msgs){ __node__.send(__msgid__,msgs);}"+
+                                 "send:function(msgs){ __node__.send(__send__,__msgid__,msgs);},"+
+                                 "done:__done__"+
                               "};\n"+
                               this.func+"\n"+
-                           "})(msg);";
+                           "})(msg,send,done);";
         this.topic = n.topic;
         this.outstandingTimers = [];
         this.outstandingIntervals = [];
@@ -104,8 +114,8 @@ module.exports = function(RED) {
                 trace: function() {
                     node.trace.apply(node, arguments);
                 },
-                send: function(id, msgs) {
+                send: function(send, id, msgs) {
-                    sendResults(node, id, msgs);
+                    sendResults(node, send, id, msgs);
                 },
                 on: function() {
                     if (arguments[0] === "input") {
@@ -223,12 +233,18 @@ module.exports = function(RED) {
                 // lineOffset: -11, // line number offset to be used for stack traces
                 // columnOffset: 0, // column number offset to be used for stack traces
             });
-            this.on("input", function(msg) {
+            this.on("input", function(msg,send,done) {
                 try {
                     var start = process.hrtime();
                     context.msg = msg;
+                    context.send = send;
+                    context.done = done;
+
                     this.script.runInContext(context);
-                    sendResults(this,msg._msgid,context.results);
+                    sendResults(this,send,msg._msgid,context.results);
+                    if (handleNodeDoneCall) {
+                        done();
+                    }
 
                     var duration = process.hrtime(start);
                     var converted = Math.floor((duration[0] * 1e9 + duration[1])/10000)/100;

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

@@ -18,10 +18,20 @@ 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;
@@ -55,6 +65,21 @@ function Node(n) {
 
 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 || [];
@@ -67,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) {
@@ -78,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);
@@ -85,6 +117,12 @@ 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
@@ -96,7 +134,15 @@ Node.prototype._complete = function(msg,error) {
     }
 }
 
+/**
+ * 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") {
@@ -115,7 +161,14 @@ Node.prototype.on = function(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") {
@@ -133,9 +186,15 @@ Node.prototype.emit = function(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,
@@ -146,6 +205,7 @@ Node.prototype._emitInput = function(arg) {
             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];
@@ -170,7 +230,50 @@ Node.prototype._emitInput = function(arg) {
     }
 }
 
+/**
+ * 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;
@@ -182,18 +285,26 @@ Node.prototype.removeAllListeners = function(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(() => {
@@ -208,6 +319,7 @@ Node.prototype.close = function(removed) {
                 })
             );
         } else {
+            // No done callback so handle synchronously
             try {
                 callback.call(node);
             } catch(err) {
@@ -230,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;
@@ -317,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 = {};
@@ -346,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 || "";
@@ -368,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) {
@@ -393,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"