fixes to event encoding and acknowledgement

lib/manager.js

@@ -329,7 +329,7 @@ Manager.prototype.handleClient = function (data, req) {
       if (count == 1) {
         // initialize the socket for all namespaces
         for (var i in self.namespaces) {
-          self.namespaces[i].socket(data.id, true);
+          self.namespaces[i].handlePacket(data.id, {type: 'connect'});
         }
 
         // handle packets for the client (all namespaces)
@@ -418,7 +418,7 @@ Manager.prototype.handleHandshake = function (data, req, res) {
   function writeErr (status, message) {
     if (data.query.jsonp) {
       res.writeHead(200);
-      res.end('io.j[' + data.query.jsonp + '](new Error(' + message + '));');
+      res.end('io.j[' + data.query.jsonp + '](new Error(' + JSON.stringify(message) + '));');
     } else {
       res.writeHead(status);
       res.end(message);
@@ -454,7 +454,7 @@ Manager.prototype.handleHandshake = function (data, req, res) {
         if (data.query.jsonp)
           hs = 'io.j[' + data.query.jsonp + '](' + JSON.stringify(hs) + ');';
 
-        res.writeHead(200);
+        res.writeHead(200, {'content-type': 'application/javascript'});
         res.end(hs);
       });
     } else {

lib/namespace.js

@@ -97,6 +97,18 @@ SocketNamespace.prototype.except = function (id) {
   return this;
 };
 
+/**
+ * Defines a filtering function which should select session ids to
+ * relay messages to (flag)
+ *
+ * @api public
+ */
+
+SocketNamespace.prototype.filter = function (fn) {
+  this.flags.filter = fn;
+  return this;
+};
+
 /**
  * Sets the default flags.
  *
@@ -107,6 +119,7 @@ SocketNamespace.prototype.setFlags = function () {
   this.flags = {
       endpoint: this.name
     , exceptions: []
+    , filter: null
   };
   return this;
 };
@@ -125,7 +138,9 @@ SocketNamespace.prototype.packet = function (packet) {
     , packet = parser.encodePacket(packet);
 
   store.clients(this.flags.endpoint, function (clients) {
-    clients.forEach(function (id) {
+    (this.flags.filter ? clients.filter(this.flags.filter) : clients)
+    .forEach(function (id) {
+      // N.B. this should be implemented as filtering function
       if (~exceptions.indexOf(id)) {
         log.debug('ignoring packet to ', id);
         return;
@@ -185,9 +200,6 @@ SocketNamespace.prototype.emit = function (name) {
 SocketNamespace.prototype.socket = function (sid, readable) {
   if (!this.sockets[sid]) {
     this.sockets[sid] = new Socket(this.manager, sid, this, readable);
-    if (this.name === '') {
-      this.emit('connection', this.sockets[sid]);
-    }
   }
 
   return this.sockets[sid];
@@ -200,7 +212,7 @@ SocketNamespace.prototype.socket = function (sid, readable) {
  */
 
 SocketNamespace.prototype.handlePacket = function (sessid, packet) {
-  var socket = this.socket(sessid)
+  var socket = this.socket(sessid, true) // readable
     , dataAck = packet.ack == 'data'
     , self = this;
 
@@ -216,13 +228,14 @@ SocketNamespace.prototype.handlePacket = function (sessid, packet) {
   switch (packet.type) {
     case 'connect':
       this.store.join(sessid, this.name, function () {
-        self.emit('connection', self.sockets[sessid]);
+        self.emit('connection', socket);
       });
       break;
 
     case 'ack':
       if (socket.acks[packet.ackId]) {
         socket.acks[packet.ackId].apply(socket, packet.args);
+        //delete socket.acks[packet.ackId];
       } else {
         this.log.info('unknown ack packet');
       }

lib/parser.js

@@ -73,7 +73,7 @@ exports.encodePacket = function (packet) {
     case 'event':
       var params = packet.args && packet.args.length
         ? JSON.stringify(packet.args) : '';
-      data = packet.name + (params !== '' ? ('\ufffd' + params) : '');
+      data = packet.name + (params !== '' ? ('\ufffc' + params) : '');
       break;
 
     case 'json':
@@ -173,7 +173,7 @@ exports.decodePacket = function (data) {
       break;
 
     case 'event':
-      var pieces = data.match(/([^\ufffd]+)(\ufffd)?(.*)/);
+      var pieces = data.match(/([^\ufffc]+)(\ufffc)?(.*)/);
       packet.name = pieces[1] || '';
       packet.args = [];
 

lib/socket.js

@@ -284,7 +284,7 @@ Socket.prototype.$emit = EventEmitter.prototype.emit;
 
 Socket.prototype.emit = function (ev) {
   if (events[ev]) {
-    return EventEmitter.prototype.emit.apply(this, arguments);
+    return this.$emit.apply(this, arguments);
   }
 
   var args = util.toArray(arguments).slice(1)

lib/stores/memory.js

@@ -263,13 +263,12 @@ Client.prototype.count = function (fn) {
  */
 
 Client.prototype.consume = function (fn) {
+  this.consumer = fn;
   this.paused = false;
 
   if (this.buffer.length) {
     fn(this.buffer, null);
     this.buffer = [];
-  } else {
-    this.consumer = fn;
   }
 
   return this;

lib/transport.js

@@ -219,12 +219,12 @@ Transport.prototype.setHeartbeatTimeout = function () {
     var self = this;
 
     this.heartbeatTimeout = setTimeout(function () {
-      self.log.debug('fired heartbeat timeout for client', self.id);
+//DVV      self.log.debug('fired heartbeat timeout for client', self.id);
       self.heartbeatTimeout = null;
       self.end(false, 'heartbeat timeout');
     }, this.manager.get('heartbeat timeout') * 1000);
 
-    this.log.debug('set heartbeat timeout for client', this.id);
+//DVV    this.log.debug('set heartbeat timeout for client', this.id);
   }
 };
 
@@ -238,7 +238,7 @@ Transport.prototype.clearHeartbeatTimeout = function () {
   if (this.heartbeatTimeout) {
     clearTimeout(this.heartbeatTimeout);
     this.heartbeatTimeout = null;
-    this.log.debug('cleared heartbeat timeout for client', this.id);
+//DVV    this.log.debug('cleared heartbeat timeout for client', this.id);
   }
 };
 
@@ -257,7 +257,7 @@ Transport.prototype.setHeartbeatInterval = function () {
       self.heartbeat();
     }, this.manager.get('heartbeat interval') * 1000);
 
-    this.log.debug('set heartbeat interval for client', this.id);
+//DVV    this.log.debug('set heartbeat interval for client', this.id);
   }
 };
 
@@ -281,7 +281,7 @@ Transport.prototype.clearTimeouts = function () {
 
 Transport.prototype.heartbeat = function () {
   if (this.open) {
-    this.log.debug('emitting heartbeat for client', this.id);
+//DVV    this.log.debug('emitting heartbeat for client', this.id);
     this.packet({ type: 'heartbeat' });
     this.setHeartbeatTimeout();
   }
@@ -298,10 +298,10 @@ Transport.prototype.heartbeat = function () {
 
 Transport.prototype.onMessage = function (packet) {
   if ('heartbeat' == packet.type) {
-    this.log.debug('got heartbeat packet');
+//DVV    this.log.debug('got heartbeat packet');
     this.store.heartbeat(this.id);
   } else if ('disconnect' == packet.type && packet.endpoint == '') {
-    this.log.debug('got disconnection packet');
+//DVV    this.log.debug('got disconnection packet');
     this.store.disconnect(this.id, true);
   } else {
     this.log.debug('got packet');
@@ -329,7 +329,7 @@ Transport.prototype.clearHeartbeatInterval = function () {
   if (this.heartbeatInterval) {
     clearTimeout(this.heartbeatInterval);
     this.heartbeatInterval = null;
-    this.log.debug('cleared heartbeat interval for client', this.id);
+//DVV    this.log.debug('cleared heartbeat interval for client', this.id);
   }
 };
 

lib/transports/htmlfile.js

@@ -51,7 +51,7 @@ HTMLFile.prototype.handleRequest = function (req) {
 
     req.res.write(
         '<html><body>'
-      + '<script>var _ = function (msg) { parent.s._(msg, document); };</script>'
+      + '<script>var _ = function (msg) { parent.s.transport._(msg, document); };</script>'
       + new Array(174).join(' ')
     );
   }
@@ -65,7 +65,9 @@ HTMLFile.prototype.handleRequest = function (req) {
  */
 
 HTMLFile.prototype.write = function (data) {
-  data = '<script>_(' + JSON.stringify(data) + ');</script>';
+  //DVV: very bad to overwrite global _, it's usually for underscore.js :|
+  // data = '<script>_(' + JSON.stringify(data) + ');</script>';
+  data = '<script>parent.s.transport._(' + JSON.stringify(data) + ', document);</script>';
 
   this.drain = false;
   this.response.write(data);