A punt at a "task socket"

squaremo · squaremo · commit 8afbe78b56c9 · 2014-04-01T22:33:50.000+01:00
This is an implementation of my example code in issue 57.
diff --git a/lib/sockets.js b/lib/sockets.js
@@ -58,6 +58,7 @@ var SOCKETS = {
   REQUEST: ReqSocket,
   REP: RepSocket,
   REPLY: RepSocket,
+  TASK: TaskSocket,
   WORKER: WorkerSocket
 };
 
@@ -154,6 +155,7 @@ function setsockopt(opt, value) {
   case 'expiration':
   case 'persistent':
   case 'topic':
+  case 'task':
     this.options[opt] = value; break;
   }
 }
@@ -471,3 +473,39 @@ RepSocket.prototype.write = function(chunk, encoding) {
 };
 
 RepSocket.prototype._read = ignore;
+
+// A task socket requires you to say which kinds of task you will
+// send, then lets you send to any one of them. This is like a PUSH
+// socket but instead of it round-robining among its connections, you
+// pick one each time (or set it for a few messages with a sockopt).
+function TaskSocket(channel, opts) {
+  Writable.call(this);
+  Socket.call(this, channel, opts);
+  this.queues = [];
+}
+inherits(TaskSocket, Writable);
+addSocketMethods(TaskSocket);
+TaskSocket.prototype.end = end;
+
+TaskSocket.prototype.connect = function(destination, callback) {
+  var queues = this.queues;
+  this.ch.assertQueue(destination,
+                 {durable: this.options.persistent})
+    .then(function(ok) {
+      queues.push(destination);
+    }).then(callback || ignore);
+};
+
+TaskSocket.prototype.write = function(chunk, encoding) {
+  return this.post(false, chunk, encoding);
+};
+
+TaskSocket.prototype.post = function(task, chunk, encoding) {
+  if (!task) task = this.options.task;
+  if (this.queues.indexOf(task) === -1)
+    throw new Error('Task "' + task + '" not connected');
+  var options = {expiration: this.options.expiration,
+                 persistent: this.options.persistent};
+  return this.ch.sendToQueue(task, bufferify(chunk, encoding),
+                             options);
+};
