Merge upstream (#135)

* Includes JustinTulloss/zeromq.node#575

* Includes JustinTulloss/zeromq.node#578

Author: n-riesco

binding.cc

@@ -434,12 +434,22 @@ namespace zmq {
       return;
     }
 
-    Local<Value> argv[3];
+    Local<Value> argv[4];
     argv[0] = Nan::New<Integer>(event_id);
     argv[1] = Nan::New<Integer>(event_value);
     argv[2] = Nan::New<String>(event_endpoint).ToLocalChecked();
+    switch (event_id) {
+      case ZMQ_EVENT_BIND_FAILED:
+      case ZMQ_EVENT_ACCEPT_FAILED:
+      case ZMQ_EVENT_CLOSE_FAILED:
+        argv[3] = ExceptionFromError();
+        break;
+      default:
+        argv[3] = Nan::Undefined();
+        break;
+    }
 
-    Nan::MakeCallback(this->handle(), callback_v.As<Function>(), 3, argv);
+    Nan::MakeCallback(this->handle(), callback_v.As<Function>(), 4, argv);
   }
 
   void

lib/index.js

@@ -530,8 +530,8 @@ Socket.prototype.monitor = function(interval, numOfEvents) {
   if (zmq.ZMQ_CAN_MONITOR) {
     var self = this;
 
-    self._zmq.onMonitorEvent = function(event_id, event_value, event_endpoint_addr) {
-      self.emit(events[event_id], event_value, event_endpoint_addr);
+    self._zmq.onMonitorEvent = function(event_id, event_value, event_endpoint_addr, ex) {
+      self.emit(events[event_id], event_value, event_endpoint_addr, ex);
     }
 
     self._zmq.onMonitorError = function(error) {
@@ -623,20 +623,26 @@ Socket.prototype.send = function(msg, flags, cb) {
   return this;
 };
 
-
-Socket.prototype._flushRead = function () {
-  var message = this._zmq.readv(); // can throw
-  if (!message) {
-    return false;
-  }
-
-  // Handle received message immediately to prevent memory leak in driver
+Socket.prototype._emitMessage = function (message) {
   if (message.length === 1) {
     // hot path
     this.emit('message', message[0]);
   } else {
     this.emit.apply(this, ['message'].concat(message));
   }
+}
+
+Socket.prototype._flushRead = function () {
+  try {
+    var message = this._zmq.readv(); // can throw
+    if (!message) {
+      return false;
+    }
+    // Handle received message immediately to prevent memory leak in driver
+    this._emitMessage(message)
+  } catch (error) {
+    this.emit('error', error); // can throw
+  }
   return true;
 };
 
@@ -669,17 +675,7 @@ Socket.prototype._flushReads = function() {
 
   this._isFlushingReads = true;
 
-  var received;
-
-  do {
-    try {
-      received = this._flushRead();
-    } catch (error) {
-      this._isFlushingReads = false;
-      this.emit('error', error); // can throw
-      return;
-    }
-  } while (received);
+  while (this._flushRead());
 
   this._isFlushingReads = false;
 

test/socket.monitor.js

@@ -48,6 +48,17 @@ describe('socket.monitor', function() {
         msg.toString().should.equal('world');
         req.close();
       });
+
+      // Test that bind errors pass an Error both to the callback
+      // and to the monitor event
+      var doubleRep = zmq.socket('rep');
+      doubleRep.monitor();
+      doubleRep.on('bind_error', function (errno, bindAddr, ex) {
+        (ex instanceof Error).should.equal(true);
+      });
+      doubleRep.bind('tcp://127.0.0.1:5423', function (error) {
+        (error instanceof Error).should.equal(true);
+      });
     });
   });
 

test/socket.pub-sub.js

@@ -91,4 +91,44 @@ describe('socket.pub-sub', function(){
     });
   });
 
+  it('should continue to deliver messages even after error in message handler is thrown', function(done){
+    var n = 0;
+
+    sub.subscribe('');
+    var errorHandlerCalled = 0;
+
+    sub.on('error', function (error) {
+      errorHandlerCalled++;
+    });
+
+    sub.on('message', function (msg) {
+      msg.should.be.an.instanceof(Buffer);
+      switch (n++) {
+        case 0:
+          msg.toString().should.equal('foo');
+          throw Error('test error');
+          break;
+        case 1:
+          msg.toString().should.equal('bar');
+          sub.close();
+          pub.close();
+          errorHandlerCalled.should.eql(1)
+          done();
+          break;
+      }
+    });
+
+    var addr = "inproc://stuff_ssps";
+
+    sub.bind(addr, function (error) {
+      if (error) throw error;
+      pub.connect(addr);
+
+      setTimeout(function() {
+        pub.send('foo');
+        pub.send('bar');
+      }, 100.0);
+    });
+  });
+
 });