This fixes burst sends on "req" sockets. (#536)

* This fixes burst sends on "req" sockets.

* Fix missing read flush triggered during send

binding.cc

@@ -1274,8 +1274,12 @@ namespace zmq {
           readsReady = true;
         }
 
-        if ((events & ZMQ_POLLOUT) == 0)
+        if ((events & ZMQ_POLLOUT) == 0) {
+          if (readsReady) {
+            socket->NotifyReadReady();
+          }
           return info.GetReturnValue().Set(false);
+        }
       }
 
       Local<Object> buf = batch->Get(i).As<Object>();

lib/index.js

@@ -682,6 +682,11 @@ Socket.prototype._flushReads = function() {
   } while (received);
 
   this._isFlushingReads = false;
+
+  // if many sends happened, but ended up in the queue (eg. in a req/rep scenario where each send must be followed by a
+  // response), we can try to send again now
+
+  this._flushWrites();
 };
 
 Socket.prototype._flushWrites = function() {

test/socket.req-rep.js

@@ -56,4 +56,40 @@ describe('socket.req-rep', function(){
     }
   });
 
+  it('should support a burst', function (done) {
+    var rep = zmq.socket('rep');
+    var req = zmq.socket('req');
+    var n = 10;
+
+    rep.on('message', function (msg) {
+      msg.should.be.an.instanceof(Buffer);
+      msg.toString().should.equal('hello');
+      rep.send('world');
+    });
+
+    rep.bind('inproc://reqrepburst', function (error) {
+      if (error) throw error;
+      req.connect('inproc://reqrepburst');
+
+      var received = 0;
+
+      req.on('message', function(msg){
+        msg.should.be.an.instanceof(Buffer);
+        msg.toString().should.equal('world');
+
+        received += 1;
+
+        if (received === n) {
+          rep.close();
+          req.close();
+          done();
+        }
+      });
+
+      for (var i = 0; i < n; i += 1) {
+        req.send('hello');
+      }
+    });
+  });
+
 });