[WIP] node-zmq 3

lib/Batch.js

@@ -0,0 +1,50 @@
+var ZMQ_SNDMORE = require('bindings')('zmq.node').sendFlags.ZMQ_SNDMORE;
+
+
+/**
+ * A batch consists of 1 or more message parts with their flags that need to be sent as one unit
+ */
+
+function Batch() {
+  this.content = [];      // buf, flags, buf, flags, ...
+  this.cbs = [];          // callbacks
+  this.isClosed = false;  // true if the last message does not have SNDMORE in its flags, false otherwise
+  this.next = null;       // next batch (for linked list of batches)
+}
+
+module.exports = Batch;
+
+
+Batch.prototype.append = function (buf, flags, cb) {
+  if (!Buffer.isBuffer(buf)) {
+    buf = new Buffer(String(buf), 'utf8');
+  }
+
+  this.content.push(buf, flags);
+
+  if (cb) {
+    this.cbs.push(cb);
+  }
+
+  if ((flags & ZMQ_SNDMORE) === 0) {
+    this.isClosed = true;
+  }
+};
+
+Batch.prototype.invokeError = function (socket, error) {
+  var returned = false;
+  for (var i = 0; i < this.cbs.length; i += 1) {
+    this.cbs[i].call(socket, error);
+    returned = true;
+  }
+
+  if (!returned) {
+    throw error;
+  }
+};
+
+Batch.prototype.invokeSent = function (socket) {
+  for (var i = 0; i < this.cbs.length; i += 1) {
+    this.cbs[i].call(socket);
+  }
+};

lib/BatchList.js

@@ -0,0 +1,54 @@
+// linked list for message batches
+
+var Batch = require('./Batch');
+
+
+function BatchList() {
+  this.firstBatch = null;
+  this.lastBatch = null;
+  this.length = 0;
+}
+
+module.exports = BatchList;
+
+
+BatchList.prototype.canSend = function () {
+  return this.firstBatch ? this.firstBatch.isClosed : false;
+};
+
+BatchList.prototype.append = function (buf, flags, cb) {
+  var batch = this.lastBatch;
+
+  if (!batch || batch.isClosed) {
+    batch = new Batch();
+
+    if (this.lastBatch) {
+      this.lastBatch.next = batch;
+    }
+
+    this.lastBatch = batch;
+
+    if (!this.firstBatch) {
+      this.firstBatch = batch;
+    }
+
+    this.length += 1;
+  }
+
+  batch.append(buf, flags, cb);
+};
+
+BatchList.prototype.fetch = function () {
+  var batch = this.firstBatch;
+  if (batch && batch.isClosed) {
+    this.firstBatch = batch.next;
+    this.length -= 1;
+    return batch;
+  }
+  return undefined;
+};
+
+BatchList.prototype.restore = function (batch) {
+  this.firstBatch = batch;
+  this.length += 1;
+};

lib/context.js

@@ -0,0 +1,69 @@
+var binding = require('bindings')('zmq.node');
+var socket = require('./socket');
+
+
+/**
+ * Map of context options in both "ZMQ_IO_THREADS" and "io_threads" style.
+ */
+
+var opts = {};
+Object.keys(binding.ctxOptions).forEach(function (name) {
+  var shortName = name.replace(/^ZMQ_/, '').toLowerCase();
+
+  opts[name] = opts[shortName] = binding.ctxOptions[name];
+});
+
+
+function Context() {
+  this._ctx = new binding.Context();
+}
+
+Context.prototype.set = function (option, value) {
+  return this._ctx.set(opts[option], value);
+};
+
+Context.prototype.get = function (option) {
+  return this._ctx.get(opts[option]);
+};
+
+Context.prototype.createSocket = function (type, options) {
+  return socket.createSocket(type, options, this);
+};
+
+Context.prototype.socket = Context.prototype.createSocket;
+
+
+exports.createContext = function (options) {
+  var context = new Context();
+
+  if (options) {
+    for (var key in options) {
+      context.set(key, options[key]);
+    }
+  }
+
+  return context;
+};
+
+
+var defaultContext;
+
+exports.getDefaultContext = function () {
+  if (!defaultContext) {
+    defaultContext = new Context();
+
+    if (process.env.ZMQ_IO_THREADS) {
+      defaultContext.set('ZMQ_IO_THREADS', parseInt(process.env.ZMQ_IO_THREADS, 10));
+    }
+  }
+
+  return defaultContext;
+};
+
+
+/**
+ * Expose the Context class and supported options
+ */
+
+exports.Context = Context;
+exports.opts = opts;