Addressing MongoAscoltatore 'close' bug

lib/mongo_ascoltatore.js

@@ -6,10 +6,10 @@ var TrieAscoltatore = require("./trie_ascoltatore");
 var AbstractAscoltatore = require('./abstract_ascoltatore');
 var SubsCounter = require("./subs_counter");
 var debug = require("debug")("ascoltatori:mongodb");
-
 var mongo = require('mongodb');
 var MongoClient = require('mongodb').MongoClient;
 var BSON = mongo.BSONPure;
+var async = require("async");
 
 /**
  * MongoAscoltatore is a class that inherits from AbstractAscoltatore.
@@ -167,28 +167,23 @@ MongoAscoltatore.prototype._poll = function(latest) {
     numberOfRetries: -1
   };
 
-  var cursor = this.collection.find({ _id: { $gte: latest } }, options);
+  this._cursor = this.collection.find({ _id: { $gte: latest } }, options);
 
   // we have created only the cursor, now take a look at the messages arrived until now
-  this._more(latest, cursor);
+  this._more(latest);
 };
 
 /**
  *
- * @param latest as above, first id to use as reference
- * @param cursor of the database
+ * @param {ObjectID} latest as above, first id to use as reference
  * @private
  */
-MongoAscoltatore.prototype._more = function(latest, cursor) {
+MongoAscoltatore.prototype._more = function(latest) {
   var that = this;
 
   debug('more');
 
-  cursor.each(that._handle(true, function(doc) {
-
-    if(that._closed) {
-      return;
-    }
+  that._cursor.each(that._handle(true, function(doc) {
 
     if (!doc) {
       debug('setting the subscriber polling');
@@ -203,7 +198,9 @@ MongoAscoltatore.prototype._more = function(latest, cursor) {
     if (doc._id > latest) {
       latest = doc._id;
       process.nextTick(function() {
-        that._ascoltatore.publish(doc.topic, doc.value);
+        if (!that._closed) {
+          that._ascoltatore.publish(doc.topic, doc.value);
+        }
       });
     }
   }));
@@ -220,13 +217,35 @@ MongoAscoltatore.prototype.unsubscribe = function() {
 MongoAscoltatore.prototype.unsub = MongoAscoltatore.prototype.unsubscribe;
 
 MongoAscoltatore.prototype.close = function close(done) {
+  var that = this;
+  var closeEmbedded = function() {
+    that._ascoltatore.close(function() {
+      that.emit("closed");
+      if (done) {
+        done();
+      }
+    });
+  };
+
   if (this._pollTimeout) {
     clearTimeout(this._pollTimeout);
   }
-  this._closed = true;
-  this._ascoltatore.close(done);
-  this.emit("closed");
-  this.db.close();
+
+  that._closed = true;
+
+  async.series([
+    function(cb) {
+      if (that._cursor) {
+        that._cursor.close(cb);
+        delete that._cursor;
+      } else {
+        cb();
+      }
+    },
+    function(cb) {
+      that.db.close(cb);
+    }
+  ], closeEmbedded);
 };
 
 util.aliasAscoltatore(MongoAscoltatore.prototype);

test/ascoltatori_spec.js

@@ -57,28 +57,4 @@ describe("ascoltatori", function() {
       a.close(done);
     });
   });
-
-
-    /*it("should create a new AbstractAscoltatore using function", function() {
-      function DummyAscoltatore(options, ascoltatori) {
-        ascoltatori.AbstractAscoltatore.call(this);
-
-        this.close = function (done) {
-          done();
-        };
-
-        this.emit("ready");
-      }
-
-      DummyAscoltatore.prototype = Object.create(ascoltatori.AbstractAscoltatore.prototype);
-
-      var a = ascoltatori.build({
-        json: false,
-        type: DummyAscoltatore
-      });
-      toClose.push(a);
-      expect(a).to.be.instanceOf(ascoltatori.AbstractAscoltatore);
-    });*/
-
-
 });

test/mongo_ascoltatore_spec.js

@@ -7,11 +7,7 @@ describe("ascoltatori.MongoAscoltatore", function() {
     this.instance.on("ready", done);
   });
 
-  afterEach(function() {
-    this.instance.close();
-  });
-
-  it("should do nothing", function(done) {
-    done();
+  afterEach(function(done) {
+    this.instance.close(done);
   });
 });