Added requeue and discard functionality to socket

christophermina · christophermina · commit bbae37a949bd · 2014-08-12T00:51:07.000-06:00
diff --git a/lib/sockets.js b/lib/sockets.js
@@ -111,7 +111,7 @@ function Socket(setup, options) {
   var ready = setup.then(function(ch) { self.ch = ch; });
   patch(self, ready,
         ['close', 'write', 'end', 'connect',
-         'setsockopt', 'ack', 'publish']);
+         'setsockopt', 'ack', 'requeue', 'discard', 'publish']);
 
   function closeAndInvalidate(event, err) {
     this.readable = this.writable = false;
@@ -347,6 +347,26 @@ WorkerSocket.prototype.ack = function() {
   this.ch.ack(msg);
 };
 
+
+WorkerSocket.prototype.requeue = function() {
+    var msg = this.unacked.shift();
+    if (!msg) {
+        throw new Error("requeue called with no unacknowledged messages");
+    }
+    this.ch.nack(msg);
+};
+
+
+WorkerSocket.prototype.discard = function() {
+    var msg = this.unacked.shift();
+    if (!msg) {
+        throw new Error("discard called with no unacknowledged messages");
+    }
+
+    this.ch.reject(msg, false);
+};
+
+
 WorkerSocket.prototype._read = ignore;
 
 
@@ -473,6 +493,27 @@ RepSocket.prototype.write = function(chunk, encoding) {
   return res;
 };
 
+
+RepSocket.prototype.requeue = function() {
+    var ch = this.ch, current = this.requests.shift();
+
+    if (!current)
+        throw new Error('Requeue with no pending request');
+
+    ch.nack(current);
+};
+
+
+RepSocket.prototype.discard = function() {
+    var ch = this.ch, current = this.requests.shift();
+
+    if (!current)
+        throw new Error('Discard with no pending request');
+
+    ch.reject(current, false);
+};
+
+
 RepSocket.prototype._read = ignore;
 
 // A task socket requires you to say which kinds of task you will
