add send callback feature + change README sample

README.md

@@ -17,7 +17,9 @@ var engine = require('engine.io')
   , server = engine.listen(80)
 
 server.on('connection', function (socket) {
-  socket.send('utf 8 string');
+  socket.send('utf 8 string', function(transport){
+  	//the message was sent successfully
+  });
 });
 ```
 

lib/socket.js

@@ -23,7 +23,8 @@ function Socket (id, server, transport) {
   this.upgraded = false;
   this.readyState = 'opening';
   this.writeBuffer = [];
-
+  this.packetSeq = 0;
+  this.packetsFn = {};
   this.setTransport(transport);
   this.onOpen();
 }
@@ -95,6 +96,8 @@ Socket.prototype.onPacket = function (packet) {
 };
 
 Socket.prototype.onError = function (err) {
+  this.packetsFn = {};
+  this.packetSeq = 0;
   debug('transport error');
   this.onClose('transport error', err);
 };
@@ -132,6 +135,8 @@ Socket.prototype.setTransport = function (transport) {
   this.transport.on('packet', this.onPacket.bind(this));
   this.transport.on('drain', this.flush.bind(this));
   this.transport.once('close', this.onClose.bind(this, 'transport close'));
+  //this function will manage packet events (also message callbacks)
+  this.setupSendCallback();
 };
 
 /**
@@ -207,22 +212,43 @@ Socket.prototype.clearTransport = function () {
 
 Socket.prototype.onClose = function (reason, description) {
   if ('closed' != this.readyState) {
+    this.packetsFn = {};
+    this.packetSeq = 0;
     this.clearTransport();
     this.readyState = 'closed';
     this.emit('close', reason, description);
   }
 };
 
+/**
+ * Setup and manage send callback
+ *
+ * @api private
+ */
+
+Socket.prototype.setupSendCallback = function () {
+  var self = this;
+  //the message was sent successfully, execute the callback 
+  this.transport.on("drain", function(seq){ 
+  	if('function' == typeof self.packetsFn[seq]) {
+      debug('executing send callback');
+      self.packetsFn[seq](self.transport);
+      delete self.packetsFn[seq];
+  	}
+  });
+};
+
 /**
  * Sends a message packet.
  *
  * @param {String} message
+ * @param {Function} callback
  * @return {Socket} for chaining
  * @api public
  */
 
-Socket.prototype.send = function (data) {
-  this.sendPacket('message', data);
+Socket.prototype.send = function (data, callback) {
+  this.sendPacket('message', data, callback);
   return this;
 };
 
@@ -234,10 +260,16 @@ Socket.prototype.send = function (data) {
  * @api private
  */
 
-Socket.prototype.sendPacket = function (type, data) {
+Socket.prototype.sendPacket = function (type, data, callback) {
   if ('closing' != this.readyState) {
     debug('sending packet "%s" (%s)', type, data);
-    this.writeBuffer.push({ type: type, data: data });
+    this.writeBuffer.push({ seq: this.packetSeq, type: type, data: data });
+    //add send callback to object
+    if('undefined' != typeof callback){
+    	this.packetsFn[this.packetSeq] = callback;
+    }
+    //increase the packet acks
+    this.packetSeq++;
     this.flush();
   }
 };

lib/transports/polling.js

@@ -21,6 +21,8 @@ module.exports = Polling;
 
 function Polling (req) {
   Transport.call(this, req);
+  //save sequence ids state
+  this.seqIds = [];
 };
 
 /**
@@ -96,7 +98,16 @@ Polling.prototype.onPollRequest = function (req, res) {
     req.on('close', onClose);
 
     this.writable = true;
-    this.emit('drain');
+    //decide to send empty `drain` or not
+    if (req.query.sid && this.writable && this.seqIds.length > 0) {
+      debug("acks received from client, emit drain");
+      var singleItem = this.seqIds.splice(0,1);
+      for (var i = 0, l = singleItem[0].length; i < l; ++i) {
+        this.emit('drain', singleItem[0][i]);
+      }
+    } else {
+      this.emit('drain');
+    }
 
     // if we're still writable but had a pending close, trigger an empty send
     if (this.writable && this.shouldClose) {
@@ -195,7 +206,13 @@ Polling.prototype.send = function (packets) {
     this.shouldClose();
     this.shouldClose = null;
   }
-
+  var pushTo = [];
+  for (var i = 0, l = packets.length; i < l; ++i) {
+    if (typeof packets[i].seq != "undefined" && packets[i].type == "message") {
+      pushTo.push(packets[i].seq);
+    }
+  }
+  if (pushTo.length > 0) this.seqIds.push(pushTo);
   this.write(parser.encodePayload(packets));
 };
 

lib/transports/websocket.js

@@ -67,15 +67,32 @@ WebSocket.prototype.onData = function (data) {
  */
 
 WebSocket.prototype.send = function (packets) {
+
+  //save packet item state in array
+  var packetItems = [];
   for (var i = 0, l = packets.length; i < l; i++) {
     var data = parser.encodePacket(packets[i]);
     debug('writing', data);
     this.writable = false;
     var self = this;
+
+    //TODO: review this part again
+    packetItems.push(packets[i]);
     this.socket.send(data, function (err){
       if (err) return self.onError('write error', err.stack);
+
+      var packetSeqId = null;
+      if(packetItems.length > 0) {
+        packetSeqId = packetItems[0].seq || null;
+        packetItems.splice(0,1);
+      }
+
       self.writable = true;
-      self.emit('drain');
+      if(packetSeqId != null) {
+      	self.emit('drain', packetSeqId);
+      } else {
+      	self.emit('drain');
+      }
     });
   }
 };

test/server.js

@@ -389,6 +389,8 @@ describe('server', function () {
       });
     });
 
+
+
     it('should arrive from server to client (multiple)', function (done) {
       var engine = listen({ allowUpgrades: false }, function (port) {
         var socket = new eioc.Socket('ws://localhost:%d'.s(port))
@@ -493,6 +495,143 @@ describe('server', function () {
     });
   });
 
+  describe('send', function(){
+    describe('callback', function(){
+         it('should execute when message sent (polling)', function (done) {
+          var engine = listen({ allowUpgrades: false }, function (port) {
+            var socket = new eioc.Socket('ws://localhost:%d'.s(port), { transports: ['polling'] }),
+                i = 0;
+            engine.on('connection', function (conn) {
+              conn.send('a', function(transport) {
+                i++;
+              });
+            });
+            socket.on('open', function () {
+              socket.on('message', function (msg) {
+                i++;
+              });
+            });
+
+            setTimeout(function(){
+                expect(i).to.be(2);
+                done();
+            },10);
+          });
+        });
+
+        it('should execute when message sent (websocket)', function (done) {
+          var engine = listen({ allowUpgrades: false }, function (port) {
+            var socket = new eioc.Socket('ws://localhost:%d'.s(port), { transports: ['websocket'] }),
+                i = 0;
+            engine.on('connection', function (conn) {
+              conn.send('a', function(transport) {
+                i++;
+              });
+            });
+            socket.on('open', function () {
+              socket.on('message', function (msg) {
+                i++;
+              });
+            });
+
+            setTimeout(function(){
+                expect(i).to.be(2);
+                done();
+            },10);
+          });
+        });
+
+        it('should execute once for each send', function (done) {
+          var engine = listen(function (port) {
+            var socket = new eioc.Socket('ws://localhost:%d'.s(port)),
+                i = 0,
+                j = 0;
+            engine.on('connection', function (conn) {
+              conn.send('b', function (transport) {
+                j++;
+              }); 
+
+              conn.send('a', function (transport) {
+                i++;
+              });
+
+            });
+            socket.on('open', function () {
+              socket.on('message', function (msg) {
+                if (msg == "a") {
+                    i++;
+                } else if (msg == "b") {
+                    j++;
+                }
+              });
+            });
+
+            setTimeout(function () {
+                expect(i).to.be(2);
+                expect(j).to.be(2);
+                done();
+            }, 50);
+          });
+        });
+
+        it('should execute in mutlipart packet', function (done) {
+          var engine = listen(function (port) {
+            var socket = new eioc.Socket('ws://localhost:%d'.s(port)),
+                i = 0;
+            engine.on('connection', function (conn) {
+              conn.send('b', function (transport) {
+                i++;
+              }); 
+
+              conn.send('a', function(transport) {
+                i++;
+              });
+
+            });
+            socket.on('open', function () {
+              socket.on('message', function (msg) {
+                i++;
+              });
+            });
+
+            setTimeout(function () {
+                expect(i).to.be(4);
+                done();
+            }, 50);
+          });
+        });
+
+        it('should execute in separate message', function (done) {
+          var engine = listen(function (port) {
+            var socket = new eioc.Socket('ws://localhost:%d'.s(port), { transports: ['websocket'] }),
+                i = 0;
+            engine.on('connection', function (conn) {
+
+              conn.send('a', function(transport) {
+                i++;
+
+                conn.send('b', function (transport) {
+                    i++;
+                }); 
+              });
+            });
+
+            socket.on('open', function () {
+              socket.on('message', function (msg) {
+                i++;
+              });
+            });
+
+            setTimeout(function () {
+                expect(i).to.be(4);
+                done();
+            }, 10);
+          });
+        });
+
+    });
+  });
+
   describe('upgrade', function () {
     it('should upgrade', function (done) {
       var engine = listen(function (port) {