bug: db.set callback doesn't fire

In node v0.10.x db.set callback would not fire on second write.
Node streams API changes meant that drain event is not emitted
in same frequency.  This fix appropriately only relies on the drain
event when the write buffer was full.

Tested in node v0.10.0 and node v0.8.17. Also makes existing tests
non version specific.

lib/dirty/dirty.js

@@ -40,7 +40,7 @@ Dirty.prototype.set = function(key, val, cb) {
     this._keys.splice(this._keys.indexOf(key), 1);
     delete this._docs[key];
   } else {
-    if (this._keys.indexOf(key) === -1) {  
+    if (this._keys.indexOf(key) === -1) {
       this._keys.push(key);
     }
     this._docs[key] = val;
@@ -177,7 +177,7 @@ Dirty.prototype._load = function() {
 
 Dirty.prototype._writeDrain = function() {
   this.flushing = false;
-  
+
   if (!this._queue.length) {
     this.emit('drain');
   } else {
@@ -224,6 +224,8 @@ Dirty.prototype._flush = function() {
     }
 
     (function(cbs) {
+      var isDrained;
+
       if (!self.path) {
         process.nextTick(function() {
           callbacks(null, cbs);
@@ -232,14 +234,19 @@ Dirty.prototype._flush = function() {
         return;
       }
 
-      self._writeStream.write(bundleStr, function(err) {
+      isDrained = self._writeStream.write(bundleStr, function(err) {
+        if (isDrained) {
+          self._writeDrain();
+        }
+
         if (!cbs.length && err) {
           self.emit('error', err);
           return;
         }
 
         callbacks(err, cbs);
       });
+
     })(cbs);
 
     bundleStr = '';

test/test-dirty.js

@@ -0,0 +1,87 @@
+var _ = require('underscore');
+var async = require('async');
+var dirty = require('../index.js');
+
+describe('db.set()', function() {
+	describe('using the memory store', function() {
+
+
+		it('should trigger the callback for each of 10 db calls', function(cb) {
+			connectToMemDb(function (count, db) {
+				db.on('error', function (err) {
+					return cb(err);
+				});
+
+				async.eachSeries(_.range(10),function (i,cb) {
+					setSample(db,cb);
+				}, cb);
+			});
+		});
+
+
+		it('should trigger the callback if provided', function(cb) {
+			connectToMemDb(function (count, db) {
+				db.on('error', function (err) {
+					return cb(err);
+				});
+				setSample(db,cb);
+			});
+		});
+	});
+
+
+	describe('using the disk store', function() {
+
+		it('should trigger the callback if provided', function(cb) {
+			connectToDiskDb(function (count, db) {
+				db.on('error', function (err) {
+					return cb(err);
+				});
+				setSample(db,cb);
+			});
+		});
+
+		it('should work again as long as a fresh connection is used', function(cb) {
+			connectToDiskDb(function (count, db) {
+				db.on('error', function (err) {
+					return cb(err);
+				});
+				setSample(db,cb);
+			});
+		});
+
+		it('callback should fire each time if set() is used multiple times over the same db connection', function(cb) {
+			connectToDiskDb(function (count, db) {
+				db.on('error', function (err) {
+					return cb(err);
+				});
+				async.eachSeries(_.range(2),function (i,cb) {
+					setSample(db, cb);
+				}, cb);
+			});
+		});
+	});
+});
+
+// Instantiate sample memory database
+function connectToMemDb(cb) {
+	var db = new(dirty.Dirty)();
+	db.on('load', function (err){
+		cb(err, db);
+	});
+}
+
+// Instantiate sample disk database
+function connectToDiskDb(cb) {
+	var db = dirty('dirty.db');
+	db.on('load', function (err) {
+		cb(err, db);
+	});
+}
+
+// Set sample db entry
+function setSample(db, cb) {
+	var key = 'sample';
+	var val = {id: Math.round(Math.random()*100)};
+	db.set(key, val, cb);
+}