Add callbacks to Message methods

Author: João Barbosa

Readme.md

@@ -151,17 +151,17 @@ Close the writer's connection(s) and fire the optional [fn] when completed.
 
  A single message.
 
-### Message#finish()
+### Message#finish([fn])
 
-  Mark message as complete.
+  Mark message as complete..
 
-### Message#requeue([delay])
+### Message#requeue([delay], [fn])
 
   Re-queue the message immediately, or with the
   given `delay` in milliseconds, or a string such
   as "5s", "10m" etc.
 
-### Message#touch()
+### Message#touch([fn])
 
   Reset the message's timeout, increasing the length
   of time before NSQD considers it timed out.

lib/connection.js

@@ -341,13 +341,17 @@ Connection.prototype.ready = function(n){
  * @api private
  */
 
-Connection.prototype.finish = function(id){
+Connection.prototype.finish = function(id, fn){
   assertValidMessageId(id);
-  var fn = this.onerror;
   var self = this;
-  if (!this._ready) return fn(new Error('cannot finish, connection not ready'));
-  this.command('FIN', [id], function(){
+  if (!this._ready) {
+    var err = new Error('cannot finish, connection not ready');
+    if (fn) fn(err);
+    return this.onerror(err);
+  }
+  this.command('FIN', [id], function(err){
     self.emit('finish', id);
+    if (fn) fn(err);
     --self.inFlight;
   });
 };
@@ -361,13 +365,17 @@ Connection.prototype.finish = function(id){
  * @api private
  */
 
-Connection.prototype.requeue = function(id, timeout){
+Connection.prototype.requeue = function(id, timeout, fn){
   assertValidMessageId(id);
-  var fn = this.onerror;
   var self = this;
-  if (!this._ready) return fn(new Error('cannot requeue, connection not ready'));
-  this.command('REQ', [id, timeout || 0], function(){
+  if (!this._ready) {
+    var err = new Error('cannot requeue, connection not ready');
+    if (fn) fn(err);
+    return this.onerror(err);
+  }
+  this.command('REQ', [id, timeout || 0], function(err){
     self.emit('requeue', id);
+    if (fn) fn(err);
     --self.inFlight;
   });
 };
@@ -379,13 +387,17 @@ Connection.prototype.requeue = function(id, timeout){
  * @api private
  */
 
-Connection.prototype.touch = function(id){
+Connection.prototype.touch = function(id, fn){
   assertValidMessageId(id);
-  var fn = this.onerror;
   var self = this;
-  if (!this._ready) return fn(new Error('cannot touch, connection not ready'));
-  this.command('TOUCH', [id], function(){
+  if (!this._ready) {
+    var err = new Error('cannot touch, connection not ready');
+    if (fn) fn(err);
+    return this.onerror(err);
+  }
+  this.command('TOUCH', [id], function(err){
     self.emit('touch', id);
+    if (fn) fn(err);
   });
 };
 

lib/message.js

@@ -41,12 +41,13 @@ function Message(body, conn) {
 /**
  * Mark the message as finished.
  *
+ * @param {Function} [fn]
  * @api public
  */
 
-Message.prototype.finish = function(){
+Message.prototype.finish = function(fn){
   this.responded = true;
-  this.conn.finish(this.id);
+  this.conn.finish(this.id, fn);
   this.trace('message:finish', { msg: this });
 };
 
@@ -77,25 +78,27 @@ Message.prototype.timedout = function(){
  * Re-queue the message with optional `delay`.
  *
  * @param {Number|String} [delay]
+ * @param {Function} [fn]
  * @api public
  */
 
-Message.prototype.requeue = function(delay){
+Message.prototype.requeue = function(delay, fn){
   if ('string' == typeof delay) delay = ms(delay);
   this.responded = true;
-  this.conn.requeue(this.id, delay);
+  this.conn.requeue(this.id, delay, fn);
   this.trace('message:requeue', { msg: this });
 };
 
 /**
  * Touch the message.
  *
+ * @param {Function} [fn]
  * @api public
  */
 
-Message.prototype.touch = function(){
+Message.prototype.touch = function(fn){
   this.lastTouch = Date.now();
-  this.conn.touch(this.id);
+  this.conn.touch(this.id, fn);
   this.trace('message:touch', { msg: this });
 };