Merge pull request #58 from LearnBoost/add/flush-drain

Add/flush drain

README.md

@@ -97,6 +97,18 @@ For more information on the client refer to the
 
 These are exposed by `require('engine.io')`:
 
+##### Events
+
+- `flush`
+    - Called when a socket buffer is being flushed.
+    - **Arguments**
+      - `Socket`: socket being flushed
+      - `Array`: write buffer
+- `drain`
+    - Called when a socket buffer is drained
+    - **Arguments**
+      - `Socket`: socket being flushed
+
 ##### Properties
 
 - `protocol` _(Number)_: protocol revision number
@@ -140,7 +152,7 @@ The main server/manager. _Inherits from EventEmitter_.
 ##### Events
 
 - `connection`
-    - Fired when a new connection is established. 
+    - Fired when a new connection is established.
     - **Arguments**
       - `Socket`: a Socket object
 
@@ -215,6 +227,12 @@ A representation of a client. _Inherits from EventEmitter_.
     - Fired when an error occurs.
     - **Arguments**
       - `Error`: error object
+- `flush`
+    - Called when the write buffer is being flushed.
+    - **Arguments**
+      - `Array`: write buffer
+- `drain`
+    - Called when the write buffer is drained
 
 ##### Properties
 

lib/socket.js

@@ -252,8 +252,12 @@ Socket.prototype.flush = function () {
   if ('closed' != this.readyState && this.transport.writable
     && this.writeBuffer.length) {
     debug('flushing buffer to transport');
+    this.emit('flush', this.writeBuffer);
+    this.server.emit('flush', this, this.writeBuffer);
     this.transport.send(this.writeBuffer);
     this.writeBuffer = [];
+    this.emit('drain');
+    this.server.emit('drain', this);
   }
 };
 
@@ -273,7 +277,7 @@ Socket.prototype.getAvailableUpgrades = function () {
     }
   }
   return availableUpgrades;
-}
+};
 
 /**
  * Closes the socket and underlying transport.

test/server.js

@@ -558,6 +558,38 @@ describe('server', function () {
         });
       });
     });
+
+    it('should trigger a flush/drain event', function(done){
+      var engine = listen({ allowUpgrades: false }, function(port){
+        engine.on('connection', function(socket){
+          var totalEvents = 4;
+
+          engine.on('flush', function(sock, buf){
+            expect(sock).to.be(socket);
+            expect(buf).to.be.an('array');
+            --totalEvents || done();
+          });
+          socket.on('flush', function(buf){
+            expect(buf).to.be.an('array');
+            --totalEvents || done();
+          });
+
+          engine.on('drain', function(sock){
+            expect(sock).to.be(socket);
+            expect(socket.writeBuffer.length).to.be(0);
+            --totalEvents || done();
+          });
+          socket.on('drain', function(){
+            expect(socket.writeBuffer.length).to.be(0);
+            --totalEvents || done();
+          });
+
+          socket.send('aaaa');
+        });
+
+        new eioc.Socket('ws://localhost:%d'.s(port));
+      });
+    });
   });
 
   describe('upgrade', function () {