Allow nodes to close asynchronously

red/nodes/Node.js

@@ -17,6 +17,7 @@
 var util = require("util");
 var EventEmitter = require("events").EventEmitter;
 var clone = require("clone");
+var when = require("when");
 
 var flows = require("./flows");
 var credentials = require('./credentials')
@@ -24,6 +25,7 @@ var comms = require("../comms");
 
 function Node(n) {
     this.id = n.id;
+    this._events = new EventEmitter();
     flows.add(this);
     this.type = n.type;
     if (n.name) {
@@ -31,13 +33,30 @@ function Node(n) {
     }
     this.wires = n.wires||[];
 }
-util.inherits(Node,EventEmitter);
 
-Node.prototype.close = function() {
-    // called when a node is removed
-    this.emit("close");
+Node.prototype.on = function(event,callback) {
+    if (event == "close") {
+        if (callback.length == 1) {
+            this.close = function() {
+                return when.promise(function(resolve) {
+                    callback(function() {
+                        resolve();
+                    });
+                });
+            }
+        } else {
+            this.close = callback;
+        }
+    }
+    this._events.on(event,callback);
 }
 
+Node.prototype.emit = function(event,args) {
+    this._events.emit(event,args);
+}
+
+Node.prototype.close = function() {
+}
 
 Node.prototype.send = function(msg) {
     // instanceof doesn't work for some reason here

red/nodes/flows.js

@@ -93,7 +93,7 @@ function stopFlows() {
     if (activeConfig&&activeConfig.length > 0) {
         util.log("[red] Stopping flows");
     }
-    flowNodes.clear();
+    return flowNodes.clear();
 }
 
 var flowNodes = module.exports = {
@@ -120,12 +120,21 @@ var flowNodes = module.exports = {
         return nodes[i];
     },
     clear: function() {
+        return when.promise(function(resolve) {
             events.emit("nodes-stopping");
+            var promises = [];
             for (var n in nodes) {
-            nodes[n].close();
+                var p = nodes[n].close();
+                if (p) {
+                    promises.push(p);
                 }
+            }
+            when.settle(promises).then(function() {
                 events.emit("nodes-stopped");
                 nodes = {};
+                resolve();
+            });
+        });
     },
     each: function(cb) {
         for (var n in nodes) {
@@ -141,9 +150,10 @@ var flowNodes = module.exports = {
     },
     setFlows: function(conf) {
         return storage.saveFlows(conf).then(function() {
-            stopFlows();
+            return stopFlows().then(function() {
                 activeConfig = conf;
                 parseConfig();
+            });
         })
     },
     stopFlows: stopFlows