added callback to client.set and producer.send

examples/producer.js

@@ -16,12 +16,18 @@ producer.on('stored', function(stored){
 
 //lets input some data
 var interval = setInterval( function(){
-    producer.send( (new Date().getTime()) + ' - New Message' );
-}, 100);
+    producer.send( (new Date().getTime()) + ' - New Message', function(err){
+      if(err){
+        console.log("ERR",err);
+      } else {
+        console.log("STORED");
+      }
+    } );
+}, 0);
 
 
 //close connection
 setTimeout( function(){
     clearInterval(interval);
     producer.close();
-}, 6000);
+}, 500);

lib/kestrelClient.js

@@ -20,6 +20,7 @@ var kestrel = function( options ){
 	}
     }
 
+    this._pendingSetCallback = null; 
 
     ee2.call(this);
 }
@@ -75,7 +76,12 @@ function _handleData(data, self){
     data = data.toString();
 
     if( data.match('STORED\r\n') != null ){
-	self.emit('stored', true);
+	    self.emit('stored', true);
+        if(self._pendingSetCallback){
+            var callback = self._pendingSetCallback;
+            self._pendingSetCallback = null;
+            callback(null);
+        }
     }
     
     if( data.match(/^VALUE/) != null ){
@@ -189,17 +195,34 @@ kestrel.prototype.close = function(){
 }
 
 
-kestrel.prototype.set = function( queue, value, lifetime ){
+kestrel.prototype.set = function( queue, value, lifetime, callback){
+    if( typeof(lifetime) === "function" ){
+    callback = lifetime;
+    lifetime = null;
+    }
     if( lifetime == undefined || lifetime == null ){
 	lifetime = 0;
     }
 
+    if(this._pendingSetCallback && callback){
+        return callback("Cannot write again. Still waiting for previous write to be stored");
+    }
+
     var command = "SET " + queue + " 0 " + lifetime + " ";
     command += Buffer.byteLength(value, 'utf8') + "\r\n" + value + "\r\n";
 
     var connection = this._getConnection();
     if( connection != null ){
-	connection.write(command);
+        if(callback){
+            this._pendingSetCallback = callback;
+        }
+    	connection.write(command);
+    } else {
+        if(callback){
+            callback("No connection");    
+        } else {
+            throw new Error("No connection");
+        }
     }
 
     return this;

lib/kestrelProducer.js

@@ -18,13 +18,17 @@ util.inherits(producer,ee2);
 
 
 
-producer.prototype.send = function(message, lifetime){
+producer.prototype.send = function(message, lifetime, callback){
+    if( typeof(lifetime) === "function" ){
+    callback = lifetime;
+    lifetime = null;
+    }
     if( lifetime == null || lifetime == undefined ){
 	lifetime = 0;
     }
     lifetime = parseInt(lifetime);
 
-    this._client.set(this._queue, message, lifetime);
+    this._client.set(this._queue, message, lifetime, callback);
 }
 
 producer.prototype.close = function(){