Release 0.7.8

Garbage collection

History.md

@@ -1,4 +1,74 @@
 
+0.7.8 / 2011-08-08
+==================
+
+  * Changed; make sure sio#listen passes options to both HTTP server and socket.io manager.
+  * Added docs for sio#listen.
+  * Added options parameter support for Manager constructor.
+  * Added memory leaks tests and test-leaks Makefile task.
+  * Removed auto npm-linking from make test.
+  * Make sure that you can disable heartbeats. [3rd-Eden]
+  * Fixed rooms memory leak [3rd-Eden]
+  * Send response once we got all POST data, not immediately [Pita]
+  * Fixed onLeave behavior with missing clientsk [3rd-Eden]
+  * Prevent duplicate references in rooms.
+  * Added alias for `to` to `in` and `in` to `to`.
+  * Fixed roomClients definition.
+  * Removed dependency on redis for installation without npm [3rd-Eden]
+  * Expose path and querystring in handshakeData [3rd-Eden]
+
+0.7.7 / 2011-07-12
+==================
+
+  * Fixed double dispatch handling with emit to closed clients.
+  * Added test for emitting to closed clients to prevent regression.
+  * Fixed race condition in redis test.
+  * Changed Transport#end instrumentation.
+  * Leveraged $emit instead of emit internally.
+  * Made tests faster.
+  * Fixed double disconnect events.
+  * Fixed disconnect logic
+  * Simplified remote events handling in Socket.
+  * Increased testcase timeout.
+  * Fixed unknown room emitting (GH-291). [3rd-Eden]
+  * Fixed `address` in handshakeData. [3rd-Eden]
+  * Removed transports definition in chat example.
+  * Fixed room cleanup
+  * Fixed; make sure the client is cleaned up after booting.
+  * Make sure to mark the client as non-open if the connection is closed.
+  * Removed unneeded `buffer` declarations.
+  * Fixed; make sure to clear socket handlers and subscriptions upon transport close.
+
+0.7.6 / 2011-06-30
+==================
+
+  * Fixed general dispatching when a client has closed.
+
+0.7.5 / 2011-06-30
+==================
+
+  * Fixed dispatching to clients that are disconnected.
+
+0.7.4 / 2011-06-30
+==================
+
+  * Fixed; only clear handlers if they were set. [level09]
+
+0.7.3 / 2011-06-30
+==================
+
+  * Exposed handshake data to clients.
+  * Refactored dispatcher interface.
+  * Changed; Moved id generation method into the manager.
+  * Added sub-namespace authorization. [3rd-Eden]
+  * Changed; normalized SocketNamespace local eventing [dvv]
+  * Changed; Use packet.reason or default to 'packet' [3rd-Eden]
+  * Changed console.error to console.log.
+  * Fixed; bind both servers at the same time do that the test never times out.
+  * Added 304 support.
+  * Removed `Transport#name` for abstract interface.
+  * Changed; lazily require http and https module only when needed. [3rd-Eden]
+
 0.7.2 / 2011-06-22
 ==================
 

Makefile

@@ -2,8 +2,8 @@
 ALL_TESTS = $(shell find test/ -name '*.test.js')
 
 run-tests:
-	@npm link > /dev/null --local
 	@./node_modules/.bin/expresso \
+		-t 3000 \
 		-I support \
 		-I lib \
 		--serial \
@@ -16,4 +16,7 @@ test:
 test-cov:
 	@TESTFLAGS=--cov $(MAKE) test
 
+test-leaks:
+	@ls test/leaks/* | xargs node --expose_debug_as=debug --expose_gc
+
 .PHONY: test

Readme.md

@@ -1,4 +1,3 @@
-
 # Socket.IO
 
 Socket.IO is a Node.JS project that makes WebSockets and realtime possible in
@@ -21,7 +20,7 @@ Next, attach it to a HTTP/HTTPS server. If you're using the fantastic `express`
 web framework:
 
 ```js
-var app = express.createServer();
+var app = express.createServer()
   , io = io.listen(app);
 
 app.listen(80);
@@ -89,7 +88,7 @@ io.sockets.on('connection', function (socket) {
   });
 
   socket.on('msg', function () {
-    socket.get('nickname', function (name) {
+    socket.get('nickname', function (err, name) {
       console.log('Chat message by ', name);
     });
   });
@@ -149,9 +148,8 @@ var news = io
 
 ```html
 <script>
-  var socket = io.connect('http://localhost/')
-    , chat = socket.of('/chat')
-    , news = socket.of('/news');
+  var chat = io.connect('http://localhost/chat')
+    , news = io.connect('http://localhost/news');
 
   chat.on('connect', function () {
     chat.emit('hi!');
@@ -277,7 +275,7 @@ Simply leverage `send` and listen on the `message` event:
 #### Server side
 
 ```js
-var io = require('socket.io-node').listen(80);
+var io = require('socket.io').listen(80);
 
 io.sockets.on('connection', function (socket) {
   socket.on('message', function () { });
@@ -307,7 +305,7 @@ Configuration in socket.io is TJ-style:
 #### Server side
 
 ```js
-var io = require('socket.io-node').listen(80);
+var io = require('socket.io').listen(80);
 
 io.configure(function () {
   io.set('transports', ['websocket', 'flashsocket', 'xhr-polling']);

examples/chat/app.js

@@ -61,14 +61,6 @@ app.listen(3000, function () {
 var io = sio.listen(app)
   , nicknames = {};
 
-io.set('transports', [
-    'websocket'
-  , 'flashsocket'
-  , 'htmlfile'
-  , 'xhr-polling'
-  , 'jsonp-polling'
-]);
-
 io.sockets.on('connection', function (socket) {
   socket.on('user message', function (msg) {
     socket.broadcast.emit('user message', socket.nickname, msg);

lib/logger.js

@@ -74,7 +74,7 @@ Logger.prototype.log = function (type) {
   if (index > this.level)
     return this;
 
-  console.error.apply(
+  console.log.apply(
       console
     , [this.colors
         ? '   \033[' + colors[index] + 'm' + pad(type) + ' -\033[39m'

lib/manager.js

@@ -1,4 +1,3 @@
-
 /*!
  * socket.io-node
  * Copyright(c) 2011 LearnBoost <dev@learnboost.com>
@@ -9,9 +8,7 @@
  * Module dependencies.
  */
 
-var http = require('http')
-  , https = require('https')
-  , fs = require('fs')
+var fs = require('fs')
   , url = require('url')
   , util = require('./util')
   , store = require('./store')
@@ -55,7 +52,7 @@ var parent = module.parent.exports
  * @api public
  */
 
-function Manager (server) {
+function Manager (server, options) {
   this.server = server;
   this.namespaces = {};
   this.sockets = this.of('');
@@ -79,8 +76,16 @@ function Manager (server) {
     , 'browser client': true
     , 'browser client minification': false
     , 'browser client etag': false
+    , 'browser client handler': false
+    , 'client store expiration': 15
   };
 
+  for (var i in options) {
+    this.settings[i] = options[i];
+  }
+
+  this.initStore();
+
   // reset listeners
   this.oldListeners = server.listeners('request');
   server.removeAllListeners('request');
@@ -95,6 +100,13 @@ function Manager (server) {
     self.handleUpgrade(req, socket, head);
   });
 
+  server.on('close', function () {
+    clearInterval(self.gc);
+  });
+
+  // run our private gc every 10 seconds
+  this.gc = setInterval(this.garbageCollection.bind(this), 10000);
+
   for (var i in transports) {
     if (transports[i].init) {
       transports[i].init(this);
@@ -216,6 +228,274 @@ Manager.prototype.configure = function (env, fn) {
   return this;
 };
 
+/**
+ * Initializes everything related to the message dispatcher.
+ *
+ * @api private
+ */
+
+Manager.prototype.initStore = function () {
+  this.handshaken = {};
+  this.connected = {};
+  this.open = {};
+  this.closed = {};
+  this.closedA = [];
+  this.rooms = {};
+  this.roomClients = {};
+
+  var self = this;
+
+  this.store.subscribe('handshake', function (id, data) {
+    self.onHandshake(id, data);
+  });
+
+  this.store.subscribe('connect', function (id) {
+    self.onConnect(id);
+  });
+
+  this.store.subscribe('open', function (id) {
+    self.onOpen(id);
+  });
+
+  this.store.subscribe('join', function (id, room) {
+    self.onJoin(id, room);
+  });
+
+  this.store.subscribe('leave', function (id, room) {
+    self.onLeave(id, room);
+  });
+
+  this.store.subscribe('close', function (id) {
+    self.onClose(id);
+  });
+
+  this.store.subscribe('dispatch', function (room, packet, volatile, exceptions) {
+    self.onDispatch(room, packet, volatile, exceptions);
+  });
+
+  this.store.subscribe('disconnect', function (id) {
+    self.onDisconnect(id);
+  });
+};
+
+/**
+ * Called when a client handshakes.
+ *
+ * @param text
+ */
+
+Manager.prototype.onHandshake = function (id, data) {
+  this.handshaken[id] = data;
+};
+
+/**
+ * Called when a client connects (ie: transport first opens)
+ *
+ * @api private
+ */
+
+Manager.prototype.onConnect = function (id) {
+  this.connected[id] = true;
+};
+
+/**
+ * Called when a client opens a request in a different node.
+ *
+ * @api private
+ */
+
+Manager.prototype.onOpen = function (id) {
+  this.open[id] = true;
+
+  // if we were buffering messages for the client, clear them
+  if (this.closed[id]) {
+    var self = this;
+
+    this.closedA.splice(this.closedA.indexOf(id), 1);
+
+    this.store.unsubscribe('dispatch:' + id, function () {
+      delete self.closed[id];
+    });
+  }
+
+  // clear the current transport
+  if (this.transports[id]) {
+    this.transports[id].discard();
+    this.transports[id] = null;
+  }
+};
+
+/**
+ * Called when a message is sent to a namespace and/or room.
+ *
+ * @api private
+ */
+
+Manager.prototype.onDispatch = function (room, packet, volatile, exceptions) {
+  if (this.rooms[room]) {
+    for (var i = 0, l = this.rooms[room].length; i < l; i++) {
+      var id = this.rooms[room][i];
+
+      if (!~exceptions.indexOf(id)) {
+        if (this.transports[id] && this.transports[id].open) {
+          this.transports[id].onDispatch(packet, volatile);
+        } else if (!volatile) {
+          this.onClientDispatch(id, packet);
+        }
+      }
+    }
+  }
+};
+
+/**
+ * Called when a client joins a nsp / room.
+ *
+ * @api private
+ */
+
+Manager.prototype.onJoin = function (id, name) {
+  if (!this.roomClients[id]) {
+    this.roomClients[id] = {};
+  }
+
+  if (!this.rooms[name]) {
+    this.rooms[name] = [];
+  }
+
+  if (!~this.rooms[name].indexOf(id)) {
+    this.rooms[name].push(id);
+    this.roomClients[id][name] = true;
+  }
+};
+
+/**
+ * Called when a client leaves a nsp / room.
+ *
+ * @param private
+ */
+
+Manager.prototype.onLeave = function (id, room) {
+  if (this.rooms[room]) {
+    var index = this.rooms[room].indexOf(id);
+
+    if (index >= 0) {
+      this.rooms[room].splice(index, 1);
+    }
+
+    if (!this.rooms[room].length) {
+      delete this.rooms[room];
+    }
+    delete this.roomClients[id][room];
+  }
+};
+
+/**
+ * Called when a client closes a request in different node.
+ *
+ * @api private
+ */
+
+Manager.prototype.onClose = function (id) {
+  if (this.open[id]) {
+    delete this.open[id];
+  }
+
+  this.closed[id] = [];
+  this.closedA.push(id);
+
+  var self = this;
+
+  this.store.subscribe('dispatch:' + id, function (packet, volatile) {
+    if (!volatile) {
+      self.onClientDispatch(id, packet);
+    }
+  });
+};
+
+/**
+ * Dispatches a message for a closed client.
+ *
+ * @api private
+ */
+
+Manager.prototype.onClientDispatch = function (id, packet) {
+  if (this.closed[id]) {
+    this.closed[id].push(packet);
+  }
+};
+
+/**
+ * Receives a message for a client.
+ *
+ * @api private
+ */
+
+Manager.prototype.onClientMessage = function (id, packet) {
+  if (this.namespaces[packet.endpoint]) {
+    this.namespaces[packet.endpoint].handlePacket(id, packet);
+  }
+};
+
+/**
+ * Fired when a client disconnects (not triggered).
+ *
+ * @api private
+ */
+
+Manager.prototype.onClientDisconnect = function (id, reason) {
+  for (var name in this.namespaces) {
+    if (this.roomClients[id][name]) {
+      this.namespaces[name].handleDisconnect(id, reason);
+    }
+  }
+
+  this.onDisconnect(id);
+};
+
+/**
+ * Called when a client disconnects.
+ *
+ * @param text
+ */
+
+Manager.prototype.onDisconnect = function (id, local) {
+  delete this.handshaken[id];
+
+  if (this.open[id]) {
+    delete this.open[id];
+  }
+
+  if (this.connected[id]) {
+    delete this.connected[id];
+  }
+
+  if (this.transports[id]) {
+    this.transports[id].discard();
+    delete this.transports[id];
+  }
+
+  if (this.closed[id]) {
+    delete this.closed[id];
+    this.closedA.splice(this.closedA.indexOf(id), 1);
+  }
+
+  if (this.roomClients[id]) {
+    for (var room in this.roomClients[id]) {
+      this.onLeave(id, room);
+    }
+    delete this.roomClients[id]
+  }
+
+  this.store.destroyClient(id, this.get('client store expiration'));
+
+  this.store.unsubscribe('dispatch:' + id);
+
+  if (local) {
+    this.store.unsubscribe('message:' + id);
+    this.store.unsubscribe('disconnect:' + id);
+  }
+};
+
 /**
  * Handles an HTTP request.
  *
@@ -302,12 +582,15 @@ Manager.prototype.handleHTTPRequest = function (data, req, res) {
 
 Manager.prototype.handleClient = function (data, req) {
   var socket = req.socket
-    , newTransport = false
+    , store = this.store
     , self = this;
 
   if (undefined != data.query.disconnect) {
-    self.log.debug('handling disconnection url');
-    self.store.disconnect(data.id, true);
+    if (this.transports[data.id] && this.transports[data.id].open) {
+      this.transports[data.id].onForcedDisconnect();
+    } else {
+      this.store.publish('disconnect-force:' + data.id);
+    }
     return;
   }
 
@@ -317,43 +600,55 @@ Manager.prototype.handleClient = function (data, req) {
     return;
   }
 
-  var transport = new transports[data.transport](this, data);
-  transport.pause();
-  transport.request = req;
+  var transport = new transports[data.transport](this, data, req)
+    , handshaken = this.handshaken[data.id];
 
-  if (!transport.open) {
-    this.log.debug('transport not writeable, not subscribing');
-    return;
-  }
+  if (handshaken) {
+    if (transport.open) {
+      if (this.closed[data.id] && this.closed[data.id].length) {
+        transport.payload(this.closed[data.id]);
+        this.closed[data.id] = [];
+      }
 
-  this.store.isHandshaken(data.id, function (err, handshaken) {
-    if (err || !handshaken) {
-      if (err) console.error(err);
-      transport.error('client not handshaken', 'reconnect');
-      return;
+      this.onOpen(data.id);
+      this.store.publish('open', data.id);
+      this.transports[data.id] = transport;
     }
 
-    self.store.client(data.id).count(function (err, count) {
-      transport.resume();
+    if (!this.connected[data.id]) {
+      this.onConnect(data.id);
+      this.store.publish('connect', data.id);
 
-      if (count == 1) {
-        // initialize the socket for all namespaces
-        for (var i in self.namespaces) {
-          var socket = self.namespaces[i].socket(data.id, true);
+      // flag as used
+      delete handshaken.issued;
+      this.onHandshake(data.id, handshaken);
+      this.store.publish('handshake', data.id, handshaken);
 
-          // echo back connect packet and fire connection event
-          if (i === '') {
-            self.namespaces[i].handlePacket(data.id, { type: 'connect' });
-          }
+      // initialize the socket for all namespaces
+      for (var i in this.namespaces) {
+        var socket = this.namespaces[i].socket(data.id, true);
+
+        // echo back connect packet and fire connection event
+        if (i === '') {
+          this.namespaces[i].handlePacket(data.id, { type: 'connect' });
         }
-
-        // handle packets for the client (all namespaces)
-        self.store.on('message:' + data.id, function (packet) {
-          self.handlePacket(data.id, packet);
-        });
       }
-    });
-  });
+
+      this.store.subscribe('message:' + data.id, function (packet) {
+        self.onClientMessage(data.id, packet);
+      });
+
+      this.store.subscribe('disconnect:' + data.id, function (reason) {
+        self.onClientDisconnect(data.id, reason);
+      });
+    }
+  } else {
+    if (transport.open) {
+      transport.error('client not handshaken', 'reconnect');
+    }
+
+    transport.discard();
+  }
 };
 
 /**
@@ -363,21 +658,24 @@ Manager.prototype.handleClient = function (data, req) {
  */
 
 Manager.static = {
-  cache:{}
-, paths: {
-      '/static/flashsocket/WebSocketMain.swf': client.dist + '/WebSocketMain.swf'
-    , '/static/flashsocket/WebSocketMainInsecure.swf': client.dist + '/WebSocketMainInsecure.swf'
-    , '/socket.io.js':  client.dist + '/socket.io.js'
-    , '/socket.io.js.min': client.dist + '/socket.io.min.js'
-  }
-, contentType: {
-      'js': 'application/javascript'
-    , 'swf': 'application/x-shockwave-flash'
-  }
-, encoding:{
-      'js': 'utf8'
-    , 'swf': 'binary'
-  }
+    cache: {}
+  , paths: {
+        '/static/flashsocket/WebSocketMain.swf': client.dist + '/WebSocketMain.swf'
+      , '/static/flashsocket/WebSocketMainInsecure.swf':
+          client.dist + '/WebSocketMainInsecure.swf'
+      , '/socket.io.js':  client.dist + '/socket.io.js'
+      , '/socket.io.js.min': client.dist + '/socket.io.min.js'
+    }
+  , mime: {
+        'js': {
+            contentType: 'application/javascript'
+          , encoding: 'utf8'
+        }
+      , 'swf': {
+            contentType: 'application/x-shockwave-flash'
+          , encoding: 'binary'
+        }
+    }
 };
 
 /**
@@ -389,15 +687,34 @@ Manager.static = {
 Manager.prototype.handleClientRequest = function (req, res, data) {
   var static = Manager.static
     , extension = data.path.split('.').pop()
-    , file = data.path + (this.enabled('browser client minification') && extension == 'js' ? '.min' : '')
+    , file = data.path + (this.enabled('browser client minification')
+        && extension == 'js' ? '.min' : '')
     , location = static.paths[file]
     , cache = static.cache[file];
 
   var self = this;
 
+  /**
+   * Writes a response, safely
+   *
+   * @api private
+   */
+
+  function write (status, headers, content, encoding) {
+    try {
+      res.writeHead(status, headers || null);
+      res.end(content || '', encoding || null);
+    } catch (e) {}
+  }
+
   function serve () {
-    var headers = {
-      'Content-Type': static.contentType[extension]
+    if (req.headers['if-none-match'] === cache.Etag) {
+      return write(304);
+    }
+
+    var mime = static.mime[extension]
+      , headers = {
+      'Content-Type': mime.contentType
     , 'Content-Length': cache.length
     };
 
@@ -405,9 +722,7 @@ Manager.prototype.handleClientRequest = function (req, res, data) {
       headers.Etag = cache.Etag;
     }
 
-    res.writeHead(200, headers);
-    res.end(cache.content, cache.encoding);
-
+    write(200, headers, cache.content, mime.encoding);
     self.log.debug('served static ' + data.path);
   }
 
@@ -416,10 +731,8 @@ Manager.prototype.handleClientRequest = function (req, res, data) {
   } else if (!cache) {
     fs.readFile(location, function (err, data) {
       if (err) {
-        res.writeHead(500);
-        res.end('Error serving socket.io client.');
-
-        self.log.warn('Can\'t cache socket.io client, ' + err.message);
+        write(500, null, 'Error serving static ' + data.path);
+        self.log.warn('Can\'t cache '+ data.path +', ' + err.message);
         return;
       }
 
@@ -427,7 +740,6 @@ Manager.prototype.handleClientRequest = function (req, res, data) {
         content: data
       , length: data.length
       , Etag: client.version
-      , encoding: static.encoding[extension]
       };
 
       serve();
@@ -437,6 +749,17 @@ Manager.prototype.handleClientRequest = function (req, res, data) {
   }
 };
 
+/**
+ * Generates a session id.
+ *
+ * @api private
+ */
+
+Manager.prototype.generateId = function () {
+  return Math.abs(Math.random() * Math.random() * Date.now() | 0).toString()
+    + Math.abs(Math.random() * Math.random() * Date.now() | 0).toString();
+};
+
 /**
  * Handles a handshake request.
  *
@@ -466,38 +789,75 @@ Manager.prototype.handleHandshake = function (data, req, res) {
     return;
   }
 
-  this.authorize(data, function (err, authorized) {
+  var handshakeData = this.handshakeData(data);
+
+  this.authorize(handshakeData, function (err, authorized, newData) {
     if (err) return error(err);
 
-    self.log.info('handshake ' + (authorized ? 'authorized' : 'unauthorized'));
-
     if (authorized) {
-      self.store.handshake(data, function (err, id) {
-        if (err) return error(err);
+      var id = self.generateId()
+        , hs = [
+              id
+            , self.enabled('heartbeats') ? self.get('heartbeat timeout') || '' : ''
+            , self.get('close timeout') || ''
+            , self.transports(data).join(',')
+          ].join(':');
 
-        var hs = [
-            id
-          , self.get('heartbeat timeout') || ''
-          , self.get('close timeout') || ''
-          , self.transports(data).join(',')
-        ].join(':');
+      if (data.query.jsonp) {
+        hs = 'io.j[' + data.query.jsonp + '](' + JSON.stringify(hs) + ');';
+        res.writeHead(200, { 'Content-Type': 'application/javascript' });
+      } else {
+        res.writeHead(200);
+      }
 
-        if (data.query.jsonp) {
-          hs = 'io.j[' + data.query.jsonp + '](' + JSON.stringify(hs) + ');';
-          res.writeHead(200, { 'Content-Type': 'application/javascript' });
-        } else {
-          res.writeHead(200);
-        }
+      res.end(hs);
 
-        res.end(hs);
-        self.log.info('handshaken', id);
-      });
+      self.onHandshake(id, newData || handshakeData);
+      self.store.publish('handshake', id, newData || handshakeData);
+
+      self.log.info('handshake authorized', id);
     } else {
       writeErr(403, 'handshake unauthorized');
+      self.log.info('handshake unauthorized');
     }
   })
 };
 
+/**
+ * Gets normalized handshake data
+ *
+ * @api private
+ */
+
+Manager.prototype.handshakeData = function (data) {
+  var connection = data.request.connection
+    , connectionAddress
+    , date = new Date;
+
+  if (connection.remoteAddress) {
+    connectionAddress = {
+        address: connection.remoteAddress
+      , port: connection.remotePort
+    }; 
+  } else if (connection.socket && connection.socket.remoteAddress) {
+    connectionAddress = {
+        address: connection.socket.remoteAddress
+      , port: connection.socket.remotePort
+    }; 
+  }
+
+  return {
+      headers: data.headers
+    , address: connectionAddress
+    , time: date.toString()
+    , query: data.query
+    , url: data.request.url
+    , xdomain: !!data.request.headers.origin
+    , secure: data.request.connection.secure
+    , issued: +date
+  };
+};
+
 /**
  * Verifies the origin of a request.
  *
@@ -624,6 +984,8 @@ Manager.prototype.checkRequest = function (req) {
 
 /**
  * Declares a socket namespace
+ *
+ * @api public
  */
 
 Manager.prototype.of = function (nsp) {
@@ -633,3 +995,26 @@ Manager.prototype.of = function (nsp) {
 
   return this.namespaces[nsp] = new SocketNamespace(this, nsp);
 };
+
+/**
+ * Perform garbage collection on long living objects and properties that cannot
+ * be removed automatically.
+ *
+ * @api private
+ */
+
+Manager.prototype.garbageCollection = function () {
+  // clean up unused handshakes
+  var ids = Object.keys(this.handshaken)
+    , i = ids.length
+    , now = Date.now()
+    , handshake;
+
+  while (i--) {
+    handshake = this.handshaken[ids[i]];
+
+    if ('issued' in handshake && (now - handshake.issued) >= 3E4) {
+      this.onDisconnect(ids[i]);
+    }
+  }
+};

lib/namespace.js

@@ -1,4 +1,3 @@
-
 /**
  * Module dependencies.
  */
@@ -24,6 +23,7 @@ function SocketNamespace (mgr, name) {
   this.manager = mgr;
   this.name = name || '';
   this.sockets = {};
+  this.auth = false;
   this.setFlags();
 };
 
@@ -33,6 +33,33 @@ function SocketNamespace (mgr, name) {
 
 SocketNamespace.prototype.__proto__ = EventEmitter.prototype;
 
+/**
+ * Copies emit since we override it
+ *
+ * @api private
+ */
+
+SocketNamespace.prototype.$emit = EventEmitter.prototype.emit;
+
+/**
+ * Retrieves all clients as Socket instances as an array.
+ *
+ * @api public
+ */
+
+SocketNamespace.prototype.clients = function (room) {
+  var room = this.name + (room !== undefined ?
+     '/' + room : '');
+
+  if (!this.manager.rooms[room]) {
+    return [];
+  }
+
+  return this.manager.rooms[room].map(function (id) {
+    return this.socket(id);
+  }, this);
+};
+
 /**
  * Access logger interface.
  *
@@ -81,8 +108,8 @@ SocketNamespace.prototype.__defineGetter__('volatile', function () {
  * @api public
  */
 
-SocketNamespace.prototype.in = function (room) {
-  this.flags.endpoint = (this.name === '' ? '' : (this.name + '/')) + room;
+SocketNamespace.prototype.in = SocketNamespace.prototype.to = function (room) {
+  this.flags.endpoint = this.name + (room ? '/' + room : '');
   return this;
 };
 
@@ -126,20 +153,8 @@ SocketNamespace.prototype.packet = function (packet) {
     , exceptions = this.flags.exceptions
     , packet = parser.encodePacket(packet);
 
-  store.clients(this.flags.endpoint, function (clients) {
-    clients.forEach(function (id) {
-      if (~exceptions.indexOf(id)) {
-        log.debug('ignoring packet to ', id);
-        return;
-      }
-
-      if (volatile) {
-        store.publish('volatile:' + id, packet);
-      } else {
-        store.client(id).publish(packet);
-      }
-    });
-  });
+  this.manager.onDispatch(this.flags.endpoint, packet, volatile, exceptions);
+  this.store.publish('dispatch', this.flags.endpoint, packet, volatile, exceptions);
 
   this.setFlags();
 
@@ -166,8 +181,8 @@ SocketNamespace.prototype.send = function (data) {
  */
 
 SocketNamespace.prototype.emit = function (name) {
-  if (name == 'connection' || name == 'newListener') {
-    return EventEmitter.prototype.emit.apply(this, arguments);
+  if (name == 'newListener') {
+    return this.$emit.apply(this, arguments);
   }
 
   return this.packet({
@@ -192,6 +207,54 @@ SocketNamespace.prototype.socket = function (sid, readable) {
   return this.sockets[sid];
 };
 
+/**
+ * Sets authorization for this namespace
+ *
+ * @api public
+ */
+
+SocketNamespace.prototype.authorization = function (fn) {
+  this.auth = fn;
+  return this;
+};
+
+/**
+ * Called when a socket disconnects entirely.
+ *
+ * @api private
+ */
+
+SocketNamespace.prototype.handleDisconnect = function (sid, reason) {
+  if (this.sockets[sid] && this.sockets[sid].readable) {
+    this.sockets[sid].onDisconnect(reason);
+    delete this.sockets[sid];
+  }
+};
+
+/**
+ * Performs authentication.
+ *
+ * @param Object client request data
+ * @api private
+ */
+
+SocketNamespace.prototype.authorize = function (data, fn) {
+  if (this.auth) {
+    var self = this;
+
+    this.auth.call(this, data, function (err, authorized) {
+      self.log.debug('client ' +
+        (authorized ? '' : 'un') + 'authorized for ' + self.name);
+      fn(err, authorized);
+    });
+  } else {
+    this.log.debug('client authorized for ' + this.name);
+    fn(null, true);
+  }
+
+  return this;
+};
+
 /**
  * Handles a packet.
  *
@@ -212,13 +275,42 @@ SocketNamespace.prototype.handlePacket = function (sessid, packet) {
     });
   };
 
+  function error (err) {
+    self.log.warn('handshake error ' + err + ' for ' + self.name);
+    socket.packet({ type: 'error', reason: err });
+  };
+
+  function connect () {
+    self.manager.onJoin(sessid, self.name);
+    self.store.publish('join', sessid, self.name);
+
+    // packet echo
+    socket.packet({ type: 'connect' });
+
+    // emit connection event
+    self.$emit('connection', socket);
+  };
+
   switch (packet.type) {
     case 'connect':
-      this.store.join(sessid, this.name, function () {
-        // packet echo
-        socket.packet({ type: 'connect' });
-        self.emit('connection', socket);
-      });
+      if (packet.endpoint == '') {
+        connect();
+      } else {
+        var manager = this.manager
+          , handshakeData = manager.handshaken[sessid];
+
+        this.authorize(handshakeData, function (err, authorized, newData) {
+          if (err) return error(err);
+
+          if (authorized) {
+            manager.onHandshake(sessid, newData || handshakeData);
+            self.store.publish('handshake', sessid, newData || handshakeData);
+            connect();
+          } else {
+            error('unauthorized');
+          }
+        });
+      }
       break;
 
     case 'ack':
@@ -239,7 +331,10 @@ SocketNamespace.prototype.handlePacket = function (sessid, packet) {
       break;
 
     case 'disconnect':
-      socket.emit('disconnect');
+      this.manager.onLeave(sessid, this.name);
+      this.store.publish('leave', sessid, this.name);
+
+      socket.$emit('disconnect', packet.reason || 'packet');
       break;
 
     case 'json':
@@ -249,6 +344,6 @@ SocketNamespace.prototype.handlePacket = function (sessid, packet) {
       if (dataAck)
         params.push(ack);
 
-      socket.emit.apply(socket, params);
+      socket.$emit.apply(socket, params);
   };
 };

lib/socket.io.js

@@ -9,15 +9,13 @@
  * Module dependencies.
  */
 
-var http = require('http')
-  , https = require('https')
-  , client = require('socket.io-client');
+var client = require('socket.io-client');
 
 /**
  * Version.
  */
 
-exports.version = '0.7.2';
+exports.version = '0.7.8';
 
 /**
  * Supported protocol version.
@@ -34,6 +32,9 @@ exports.clientVersion = client.version;
 /**
  * Attaches a manager
  *
+ * @param {HTTPServer/Number} a HTTP/S server or a port number to listen on.
+ * @param {Object} opts to be passed to Manager and/or http server
+ * @param {Function} callback if a port is supplied
  * @api public
  */
 
@@ -53,9 +54,9 @@ exports.listen = function (server, options, fn) {
     var port = server;
 
     if (options && options.key)
-      server = https.createServer(options, server);
+      server = require('https').createServer(options);
     else
-      server = http.createServer();
+      server = require('http').createServer();
 
     // default response
     server.on('request', function (req, res) {
@@ -67,7 +68,7 @@ exports.listen = function (server, options, fn) {
   }
 
   // otherwise assume a http/s server
-  return new exports.Manager(server);
+  return new exports.Manager(server, options);
 };
 
 /**
@@ -102,6 +103,22 @@ exports.Socket = require('./socket');
 
 exports.Store = require('./store');
 
+/**
+ * Memory Store constructor.
+ *
+ * @api public
+ */
+
+exports.MemoryStore = require('./stores/memory');
+
+/**
+ * Redis Store constructor.
+ *
+ * @api public
+ */
+
+exports.RedisStore = require('./stores/redis');
+
 /**
  * Parser.
  *

lib/socket.js

@@ -19,22 +19,6 @@ var parser = require('./parser')
 
 exports = module.exports = Socket;
 
-/**
- * Reserved event names.
- */
-
-var events = {
-    message: 1
-  , connect: 1
-  , disconnect: 1
-  , open: 1
-  , close: 1
-  , error: 1
-  , retry: 1
-  , reconnect: 1
-  , newListener: 1
-};
-
 /**
  * Socket constructor.
  *
@@ -53,14 +37,8 @@ function Socket (manager, id, nsp, readable) {
   this.ackPackets = 0;
   this.acks = {};
   this.setFlags();
-
-  if (readable) {
-    var self = this;
-
-    this.store.once('disconnect:' + id, function (reason) {
-      self.onDisconnect(reason);
-    });
-  }
+  this.readable = readable;
+  this.store = this.manager.store.client(this.id);
 };
 
 /**
@@ -70,13 +48,13 @@ function Socket (manager, id, nsp, readable) {
 Socket.prototype.__proto__ = EventEmitter.prototype;
 
 /**
- * Accessor shortcut for the store.
+ * Accessor shortcut for the handshake data
  *
  * @api private
  */
 
-Socket.prototype.__defineGetter__('store', function () {
-  return this.manager.store;
+Socket.prototype.__defineGetter__('handshake', function () {
+  return this.manager.handshaken[this.id];
 });
 
 /**
@@ -128,7 +106,7 @@ Socket.prototype.__defineGetter__('broadcast', function () {
  * @api public
  */
 
-Socket.prototype.to = function (room) {
+Socket.prototype.to = Socket.prototype.in = function (room) {
   this.flags.room = room;
   return this;
 };
@@ -155,7 +133,7 @@ Socket.prototype.setFlags = function () {
 
 Socket.prototype.onDisconnect = function (reason) {
   if (!this.disconnected) {
-    this.emit('disconnect', reason);
+    this.$emit('disconnect', reason);
     this.disconnected = true;
   }
 };
@@ -167,8 +145,17 @@ Socket.prototype.onDisconnect = function (reason) {
  */
 
 Socket.prototype.join = function (name, fn) {
-  var nsp = this.namespace.name;
-  this.store.join(this.id, (nsp === '' ? '' : (nsp + '/')) + name, fn);
+  var nsp = this.namespace.name
+    , name = (nsp + '/') + name;
+
+  this.manager.onJoin(this.id, name);
+  this.manager.store.publish('join', this.id, name);
+
+  if (fn) {
+    this.log.warn('Client#join callback is deprecated');
+    fn();
+  }
+
   return this;
 };
 
@@ -179,8 +166,17 @@ Socket.prototype.join = function (name, fn) {
  */
 
 Socket.prototype.leave = function (name, fn) {
-  var nsp = this.namespace.name;
-  this.store.leave(this.id, (nsp === '' ? '' : (nsp + '/')) + name, fn);
+  var nsp = this.namespace.name
+    , name = (nsp + '/') + name;
+
+  this.manager.onLeave(this.id, name);
+  this.manager.store.publish('leave', this.id, name);
+
+  if (fn) {
+    this.log.warn('Client#leave callback is deprecated');
+    fn();
+  }
+
   return this;
 };
 
@@ -198,11 +194,7 @@ Socket.prototype.packet = function (packet) {
     packet.endpoint = this.flags.endpoint;
     packet = parser.encodePacket(packet);
 
-    if (this.flags.volatile) {
-      this.store.publish('volatile:' + this.id, packet);
-    } else {
-      this.store.client(this.id).publish(packet);
-    }
+    this.dispatch(packet, this.flags.volatile);
   }
 
   this.setFlags();
@@ -210,6 +202,24 @@ Socket.prototype.packet = function (packet) {
   return this;
 };
 
+/**
+ * Dispatches a packet
+ *
+ * @api private
+ */
+
+Socket.prototype.dispatch = function (packet, volatile) {
+  if (this.manager.transports[this.id] && this.manager.transports[this.id].open) {
+    this.manager.transports[this.id].onDispatch(packet, volatile);
+  } else {
+    if (!volatile) {
+      this.manager.onClientDispatch(this.id, packet, volatile);
+    }
+
+    this.manager.store.publish('dispatch:' + this.id, packet, volatile);
+  }
+};
+
 /**
  * Stores data for the client.
  *
@@ -217,7 +227,7 @@ Socket.prototype.packet = function (packet) {
  */
 
 Socket.prototype.set = function (key, value, fn) {
-  this.store.client(this.id).set(key, value, fn);
+  this.store.set(key, value, fn);
   return this;
 };
 
@@ -228,7 +238,29 @@ Socket.prototype.set = function (key, value, fn) {
  */
 
 Socket.prototype.get = function (key, fn) {
-  this.store.client(this.id).get(key, fn);
+  this.store.get(key, fn);
+  return this;
+};
+
+/**
+ * Checks data for the client
+ *
+ * @api public
+ */
+
+Socket.prototype.has = function (key, fn) {
+  this.store.has(key, fn);
+  return this;
+};
+
+/**
+ * Deletes data for the client
+ *
+ * @api public
+ */
+
+Socket.prototype.del = function (key, fn) {
+  this.store.del(key, fn);
   return this;
 };
 
@@ -241,7 +273,13 @@ Socket.prototype.get = function (key, fn) {
 Socket.prototype.disconnect = function () {
   if (!this.disconnected) {
     this.log.info('booting client');
-    this.store.disconnect(this.id, true);
+
+    if (this.manager.transports[this.id] && this.manager.transports[this.id].open) {
+      this.manager.transports[this.id].onForcedDisconnect();
+    } else {
+      this.manager.onClientDisconnect(this.id);
+      this.manager.store.publish('disconnect:' + this.id);
+    }
   }
 
   return this;
@@ -283,7 +321,7 @@ Socket.prototype.$emit = EventEmitter.prototype.emit;
  */
 
 Socket.prototype.emit = function (ev) {
-  if (events[ev]) {
+  if (ev == 'newListener') {
     return this.$emit.apply(this, arguments);
   }
 

lib/store.js

@@ -5,41 +5,86 @@
  * MIT Licensed
  */
 
-/**
- * Module dependencies.
- */
-
-var EventEmitter = process.EventEmitter;
-
 /**
  * Expose the constructor.
  */
 
 exports = module.exports = Store;
 
+/**
+ * Module dependencies.
+ */
+
+var EventEmitter = process.EventEmitter;
+
 /**
  * Store interface
  *
  * @api public
  */
 
-function Store () {};
+function Store (options) {
+  this.options = options;
+  this.clients = {};
+};
 
 /**
- * Inherits from EventEmitter
+ * Inherit from EventEmitter.
  */
 
 Store.prototype.__proto__ = EventEmitter.prototype;
 
 /**
- * Log accessor.
+ * Initializes a client store
  *
+ * @param {String} id
  * @api public
  */
 
-Store.prototype.__defineGetter__('log', function () {
-  return this.manager.log;
-});
+Store.prototype.client = function (id) {
+  if (!this.clients[id]) {
+    this.clients[id] = new (this.constructor.Client)(this, id);
+  }
+
+  return this.clients[id];
+};
+
+/**
+ * Destroys a client
+ *
+ * @api {String} sid
+ * @param {Number} number of seconds to expire client data
+ * @api private
+ */
+
+Store.prototype.destroyClient = function (id, expiration) {
+  if (this.clients[id]) {
+    this.clients[id].destroy(expiration);
+    delete this.clients[id];
+  }
+
+  return this;
+};
+
+/**
+ * Destroys the store
+ *
+ * @param {Number} number of seconds to expire client data
+ * @api private
+ */
+
+Store.prototype.destroy = function (clientExpiration) {
+  var keys = Object.keys(this.clients)
+    , count = keys.length;
+
+  for (var i = 0, l = count; i < l; i++) {
+    this.destroyClient(keys[i], clientExpiration);
+  }
+
+  this.clients = {};
+
+  return this;
+};
 
 /**
  * Client.
@@ -50,6 +95,4 @@ Store.prototype.__defineGetter__('log', function () {
 Store.Client = function (store, id) {
   this.store = store;
   this.id = id;
-  this.buffer = [];
-  this.dict = {};
 };

lib/stores/memory.js

@@ -10,7 +10,7 @@
  */
 
 var crypto = require('crypto')
-  , Store = require('../store')
+  , Store = require('../store');
 
 /**
  * Exports the constructor.
@@ -26,9 +26,7 @@ Memory.Client = Client;
  */
 
 function Memory (opts) {
-  this.handshaken = [];
-  this.clientsMap = {};
-  this.rooms = {};
+  Store.call(this, opts);
 };
 
 /**
@@ -38,192 +36,28 @@ function Memory (opts) {
 Memory.prototype.__proto__ = Store.prototype;
 
 /**
- * Handshake a client.
- *
- * @param {Object} client request object
- * @param {Function} callback
- * @api public
- */
-
-Memory.prototype.handshake = function (data, fn) {
-  var id = this.generateId();
-  this.handshaken.push(id);
-  fn(null, id);
-  return this;
-};
-
-/**
- * Checks if a client is handshaken.
- *
- * @api public
- */
-
-Memory.prototype.isHandshaken = function (id, fn) {
-  fn(null, ~this.handshaken.indexOf(id));
-  return this;
-};
-
-/**
- * Generates a random id.
+ * Publishes a message.
  *
  * @api private
  */
 
-Memory.prototype.generateId = function () {
-  var rand = String(Math.random() * Math.random() * Date.now());
-  return crypto.createHash('md5').update(rand).digest('hex');
-};
+Memory.prototype.publish = function () { };
 
 /**
- * Retrieves a client store instance.
- *
- * @api public
- */
-
-Memory.prototype.client = function (id) {
-  if (!this.clientsMap[id]) {
-    this.clientsMap[id] = new Memory.Client(this, id);
-    this.log.debug('initializing client store for', id);
-  }
-
-  return this.clientsMap[id];
-};
-
-/**
- * Called when a client disconnects.
- *
- * @api public
- */
-
-Memory.prototype.disconnect = function (id, force, reason) {
-  if (~this.handshaken.indexOf(id)) {
-    this.log.debug('destroying dispatcher for', id);
-
-    this.handshaken.splice(this.handshaken.indexOf(id), 1);
-    this.clientsMap[id].destroy();
-    this.clientsMap[id] = null;
-
-    if (force)
-      this.publish('disconnect-force:' + id, reason);
-
-    this.publish('disconnect:' + id, reason);
-  }
-
-  return this;
-};
-
-/**
- * Relays a heartbeat message.
+ * Subscribes to a channel
  *
  * @api private
  */
 
-Memory.prototype.heartbeat = function (id) {
-  return this.publish('heartbeat-clear:' + id);
-};
+Memory.prototype.subscribe = function () { };
 
 /**
- * Relays a packet
+ * Unsubscribes
  *
  * @api private
  */
 
-Memory.prototype.message = function (id, packet) {
-  return this.publish('message:' + id, packet);
-};
-
-/**
- * Returns client ids in a particular room
- *
- * @api public
- */
-
-Memory.prototype.clients = function (room, fn) {
-  if ('function' == typeof room) {
-    fn = room;
-    room = '';
-  }
-
-  fn && fn(this.rooms[room] || []);
-};
-
-/**
- * Joins a user to a room
- *
- * @api private
- */
-
-Memory.prototype.join = function (sid, room, fn) {
-  if (!this.rooms[room]) {
-    this.rooms[room] = [];
-  }
-
-  this.client(sid).rooms[room] = this.rooms[room].length;
-  this.rooms[room].push(sid);
-
-  fn && fn();
-
-  return this;
-};
-
-/**
- * Removes a user from a room
- *
- * @api private
- */
-
-Memory.prototype.leave = function (sid, room, fn) {
-  if (!this.rooms[room] || this.client(sid).rooms[room] === undefined) {
-    return this;
-  }
-
-  var i = this.client(sid).rooms[room];
-  this.rooms[room][i] = null;
-  delete this.client(sid).rooms[room];
-
-  fn && fn();
-
-  return this;
-};
-
-/**
- * Simple publish
- *
- * @api public
- */
-
-Memory.prototype.publish = function (ev, data, fn) {
-  if ('function' == typeof data) {
-    fn = data;
-    data = undefined;
-  }
-
-  this.emit(ev, data);
-  if (fn) fn();
-
-  return this;
-};
-
-/**
- * Simple subscribe
- *
- * @api public
- */
-
-Memory.prototype.subscribe = function (chn, fn) {
-  this.on(chn, fn);
-  return this;
-};
-
-/**
- * Simple unsubscribe
- *
- * @api public
- */
-
-Memory.prototype.unsubscribe = function (chn) {
-  this.removeAllListeners(chn);
-};
+Memory.prototype.unsubscribe = function () { };
 
 /**
  * Client constructor
@@ -233,9 +67,7 @@ Memory.prototype.unsubscribe = function (chn) {
 
 function Client () {
   Store.Client.apply(this, arguments);
-  this.reqs = 0;
-  this.paused = true;
-  this.rooms = {};
+  this.data = {};
 };
 
 /**
@@ -244,73 +76,6 @@ function Client () {
 
 Client.prototype.__proto__ = Store.Client;
 
-/**
- * Counts transport requests.
- *
- * @api public
- */
-
-Client.prototype.count = function (fn) {
-  fn(null, ++this.reqs);
-  return this;
-};
-
-/**
- * Sets up queue consumption
- *
- * @api public
- */
-
-Client.prototype.consume = function (fn) {
-  this.consumer = fn;
-  this.paused = false;
-
-  if (this.buffer.length) {
-    fn(this.buffer, null);
-    this.buffer = [];
-  }
-
-  return this;
-};
-
-/**
- * Publishes a message to be sent to the client.
- *
- * @String encoded message
- * @api public
- */
-
-Client.prototype.publish = function (msg) {
-  if (this.paused) {
-    this.buffer.push(msg);
-  } else {
-    this.consumer(null, msg);
-  }
-
-  return this;
-};
-
-/**
- * Pauses the stream.
- *
- * @api public
- */
-
-Client.prototype.pause = function () {
-  this.paused = true;
-  return this;
-};
-
-/**
- * Destroys the client.
- *
- * @api public
- */
-
-Client.prototype.destroy = function () {
-  this.buffer = null;
-};
-
 /**
  * Gets a key
  *
@@ -318,7 +83,7 @@ Client.prototype.destroy = function () {
  */
 
 Client.prototype.get = function (key, fn) {
-  fn(null, this.dict[key]);
+  fn(null, this.data[key] === undefined ? null : this.data[key]);
   return this;
 };
 
@@ -329,18 +94,50 @@ Client.prototype.get = function (key, fn) {
  */
 
 Client.prototype.set = function (key, value, fn) {
-  this.dict[key] = value;
+  this.data[key] = value;
   fn && fn(null);
   return this;
 };
 
 /**
- * Emits a message incoming from client.
+ * Has a key
  *
+ * @api public
+ */
+
+Client.prototype.has = function (key, fn) {
+  fn(null, key in this.data);
+};
+
+/**
+ * Deletes a key
+ *
+ * @api public
+ */
+
+Client.prototype.del = function (key, fn) {
+  delete this.data[key];
+  fn && fn(null);
+  return this;
+};
+
+/**
+ * Destroys the client.
+ *
+ * @param {Number} number of seconds to expire data
  * @api private
  */
 
-Client.prototype.onMessage = function (msg) {
-  this.store.emit('message:' + id, msg);
-};
+Client.prototype.destroy = function (expiration) {
+  if ('number' != typeof expiration) {
+    this.data = {};
+  } else {
+    var self = this;
 
+    setTimeout(function () {
+      self.data = {};
+    }, expiration * 1000);
+  }
+
+  return this;
+};

lib/stores/redis.js

@@ -0,0 +1,250 @@
+
+/*!
+ * socket.io-node
+ * Copyright(c) 2011 LearnBoost <dev@learnboost.com>
+ * MIT Licensed
+ */
+
+/**
+ * Module dependencies.
+ */
+
+var crypto = require('crypto')
+  , Store = require('../store')
+  , assert = require('assert');
+
+/**
+ * Exports the constructor.
+ */
+
+exports = module.exports = Redis;
+Redis.Client = Client;
+
+/**
+ * Redis store.
+ * Options:
+ *     - nodeId (fn) gets an id that uniquely identifies this node
+ *     - redis (fn) redis constructor, defaults to redis
+ *     - redisPub (object) options to pass to the pub redis client
+ *     - redisSub (object) options to pass to the sub redis client
+ *     - redisClient (object) options to pass to the general redis client
+ *     - pack (fn) custom packing, defaults to JSON or msgpack if installed
+ *     - unpack (fn) custom packing, defaults to JSON or msgpack if installed
+ *
+ * @api public
+ */
+
+function Redis (opts) {
+  opts = opts || {};
+
+  // node id to uniquely identify this node
+  var nodeId = opts.nodeId || function () {
+    // by default, we generate a random id 
+    return Math.abs(Math.random() * Math.random() * Date.now() | 0);
+  };
+
+  this.nodeId = nodeId();
+
+  // packing / unpacking mechanism
+  if (opts.pack) {
+    this.pack = opts.pack;
+    this.unpack = opts.unpack;
+  } else {
+    try {
+      var msgpack = require('msgpack');
+      this.pack = msgpack.pack;
+      this.unpack = msgpack.unpack;
+    } catch (e) {
+      this.pack = JSON.stringify;
+      this.unpack = JSON.parse;
+    }
+  }
+
+  var redis = opts.redis || require('redis');
+
+  // initialize a pubsub client and a regular client
+  this.pub = redis.createClient(opts.redisPub);
+  this.sub = redis.createClient(opts.redisSub);
+  this.cmd = redis.createClient(opts.redisClient);
+
+  Store.call(this, opts);
+};
+
+/**
+ * Inherits from Store.
+ */
+
+Redis.prototype.__proto__ = Store.prototype;
+
+/**
+ * Publishes a message.
+ *
+ * @api private
+ */
+
+Redis.prototype.publish = function (name) {
+  var args = Array.prototype.slice.call(arguments, 1);
+  this.pub.publish(name, this.pack({ nodeId: this.nodeId, args: args }));
+  this.emit.apply(this, ['publish', name].concat(args));
+};
+
+/**
+ * Subscribes to a channel
+ *
+ * @api private
+ */
+
+Redis.prototype.subscribe = function (name, consumer, fn) {
+  this.sub.subscribe(name);
+
+  if (consumer || fn) {
+    var self = this;
+
+    self.sub.on('subscribe', function subscribe (ch) {
+      if (name == ch) {
+        function message (ch, msg) {
+          if (name == ch) {
+            msg = self.unpack(msg);
+
+            // we check that the message consumed wasnt emitted by this node
+            if (self.nodeId != msg.nodeId) {
+              consumer.apply(null, msg.args);
+            }
+          }
+        };
+
+        self.sub.on('message', message);
+
+        self.on('unsubscribe', function unsubscribe (ch) {
+          if (name == ch) {
+            self.sub.removeListener('message', message);
+            self.removeEvent('unsubscribe', unsubscribe);
+          }
+        });
+
+        self.sub.removeListener('subscribe', subscribe);
+
+        fn && fn();
+      }
+    });
+  }
+
+  this.emit('subscribe', name, consumer, fn);
+};
+
+/**
+ * Unsubscribes
+ *
+ * @api private
+ */
+
+Redis.prototype.unsubscribe = function (name, fn) {
+  this.sub.unsubscribe(name);
+
+  if (fn) {
+    var client = this.sub;
+
+    client.on('unsubscribe', function unsubscribe (ch) {
+      if (name == ch) {
+        fn();
+        client.removeListener('unsubscribe', unsubscribe);
+      }
+    });
+  }
+
+  this.emit('unsubscribe', name, fn);
+};
+
+/**
+ * Destroys the store
+ *
+ * @api public
+ */
+
+Redis.prototype.destroy = function () {
+  Store.prototype.destroy.call(this);
+
+  this.pub.end();
+  this.sub.end();
+  this.cmd.end();
+};
+
+/**
+ * Client constructor
+ *
+ * @api private
+ */
+
+function Client (store, id) {
+  Store.Client.call(this, store, id);
+};
+
+/**
+ * Inherits from Store.Client
+ */
+
+Client.prototype.__proto__ = Store.Client;
+
+/**
+ * Redis hash get
+ *
+ * @api private
+ */
+
+Client.prototype.get = function (key, fn) {
+  this.store.cmd.hget(this.id, key, fn);
+  return this;
+};
+
+/**
+ * Redis hash set
+ *
+ * @api private
+ */
+
+Client.prototype.set = function (key, value, fn) {
+  this.store.cmd.hset(this.id, key, value, fn);
+  return this;
+};
+
+/**
+ * Redis hash del
+ *
+ * @api private
+ */
+
+Client.prototype.del = function (key, fn) {
+  this.store.cmd.hdel(this.id, key, fn);
+  return this;
+};
+
+/**
+ * Redis hash has
+ *
+ * @api private
+ */
+
+Client.prototype.has = function (key, fn) {
+  this.store.cmd.hexists(this.id, key, function (err, has) {
+    if (err) return fn(err);
+    fn(null, !!has);
+  });
+  return this;
+};
+
+/**
+ * Destroys client
+ *
+ * @param {Number} number of seconds to expire data
+ * @api private
+ */
+
+Client.prototype.destroy = function (expiration) {
+  if ('number' != typeof expiration) {
+    this.store.cmd.del(this.id);
+  } else {
+    this.store.cmd.expire(this.id, expiration);
+  }
+
+  return this;
+};

lib/transport.js

@@ -23,22 +23,13 @@ exports = module.exports = Transport;
  * @api public
  */
 
-function Transport (mng, data) {
+function Transport (mng, data, req) {
   this.manager = mng;
   this.id = data.id;
-  this.paused = true;
   this.disconnected = false;
   this.drained = true;
-};
-
-/**
- * Sets the corresponding request object.
- */
-
-Transport.prototype.__defineSetter__('request', function (req) {
-  this.log.debug('setting request', req.method, req.url);
   this.handleRequest(req);
-});
+};
 
 /**
  * Access the logger.
@@ -67,29 +58,14 @@ Transport.prototype.__defineGetter__('store', function () {
  */
 
 Transport.prototype.handleRequest = function (req) {
+  this.log.debug('setting request', req.method, req.url);
   this.req = req;
 
   if (req.method == 'GET') {
     this.socket = req.socket;
     this.open = true;
     this.drained = true;
-
-    this.log.debug('publishing that', this.id, 'connected');
-
-    var self = this;
-
-    this.store.publish('open:' + this.id, function () {
-      self.store.once('open:' + self.id, function () {
-        self.log.info('request for existing session connection change');
-        self.close();
-        self.clearTimeouts();
-        self.clearHandlers();
-      });
-
-      if (!self.paused) {
-        self.subscribe();
-      }
-    });
+    this.setHeartbeatInterval();
 
     this.setHandlers();
     this.onSocketConnect();
@@ -113,17 +89,18 @@ Transport.prototype.onSocketConnect = function () { };
 Transport.prototype.setHandlers = function () {
   var self = this;
 
-  this.store.once('disconnect-force:' + this.id, function () {
+  // we need to do this in a pub/sub way since the client can POST the message
+  // over a different socket (ie: different Transport instance)
+  this.store.subscribe('heartbeat-clear:' + this.id, function () {
+    self.onHeartbeatClear();
+  });
+
+  this.store.subscribe('disconnect-force:' + this.id, function () {
     self.onForcedDisconnect();
   });
 
-  this.store.on('heartbeat-clear:' + this.id, function () {
-    self.clearHeartbeatTimeout();
-    self.setHeartbeatInterval();
-  });
-
-  this.store.on('volatile:' + this.id, function (packet) {
-    self.writeVolatile(packet);
+  this.store.subscribe('dispatch:' + this.id, function (packet, volatile) {
+    self.onDispatch(packet, volatile);
   });
 
   this.bound = {
@@ -137,6 +114,8 @@ Transport.prototype.setHandlers = function () {
   this.socket.on('close', this.bound.close);
   this.socket.on('error', this.bound.error);
   this.socket.on('drain', this.bound.drain);
+
+  this.handlersSet = true;
 };
 
 /**
@@ -146,14 +125,16 @@ Transport.prototype.setHandlers = function () {
  */
 
 Transport.prototype.clearHandlers = function () {
-  this.store.unsubscribe('disconnect-force:' + this.id);
-  this.store.unsubscribe('heartbeat-clear:' + this.id);
-  this.store.unsubscribe('volatile:' + this.id);
+  if (this.handlersSet) {
+    this.store.unsubscribe('disconnect-force:' + this.id);
+    this.store.unsubscribe('heartbeat-clear:' + this.id);
+    this.store.unsubscribe('dispatch:' + this.id);
 
-  this.socket.removeListener('end', this.bound.end);
-  this.socket.removeListener('close', this.bound.close);
-  this.socket.removeListener('error', this.bound.error);
-  this.socket.removeListener('drain', this.bound.drain);
+    this.socket.removeListener('end', this.bound.end);
+    this.socket.removeListener('close', this.bound.close);
+    this.socket.removeListener('error', this.bound.error);
+    this.socket.removeListener('drain', this.bound.drain);
+  }
 };
 
 /**
@@ -163,10 +144,7 @@ Transport.prototype.clearHandlers = function () {
  */
 
 Transport.prototype.onSocketEnd = function () {
-  // we check that the socket wasn't swapped
-  // we don't want to sever a connection that's not active, since we don't kill
-  // inactive sockets that the browser might reuse for other purposes
-  this.end(false, 'socket end');
+  this.end('socket end');
 };
 
 /**
@@ -176,7 +154,7 @@ Transport.prototype.onSocketEnd = function () {
  */
 
 Transport.prototype.onSocketClose = function (error) {
-  this.end(false, error ? 'socket error' : 'socket close');
+  this.end(error ? 'socket error' : 'socket close');
 };
 
 /**
@@ -204,6 +182,17 @@ Transport.prototype.onSocketDrain = function () {
   this.drained = true;
 };
 
+/**
+ * Called upon receiving a heartbeat packet.
+ *
+ * @api private
+ */
+
+Transport.prototype.onHeartbeatClear = function () {
+  this.clearHeartbeatTimeout();
+  this.setHeartbeatInterval();
+};
+
 /**
  * Called upon a forced disconnection.
  *
@@ -216,7 +205,21 @@ Transport.prototype.onForcedDisconnect = function () {
     if (this.open) {
       this.packet({ type: 'disconnect' });
     }
-    this.end(true);
+    this.end('booted');
+  }
+};
+
+/**
+ * Dispatches a packet.
+ *
+ * @api private
+ */
+
+Transport.prototype.onDispatch = function (packet, volatile) {
+  if (volatile) {
+    this.writeVolatile(packet);
+  } else {
+    this.write(packet);
   }
 };
 
@@ -231,7 +234,7 @@ Transport.prototype.setCloseTimeout = function () {
     this.closeTimeout = setTimeout(function () {
       self.log.debug('fired close timeout for client', self.id);
       self.closeTimeout = null;
-      self.end(false, 'close timeout');
+      self.end('close timeout');
     }, this.manager.get('close timeout') * 1000);
 
     this.log.debug('set close timeout for client', this.id);
@@ -256,13 +259,13 @@ Transport.prototype.clearCloseTimeout = function () {
  */
 
 Transport.prototype.setHeartbeatTimeout = function () {
-  if (!this.heartbeatTimeout) {
+  if (!this.heartbeatTimeout && this.manager.enabled('heartbeats')) {
     var self = this;
 
     this.heartbeatTimeout = setTimeout(function () {
       self.log.debug('fired heartbeat timeout for client', self.id);
       self.heartbeatTimeout = null;
-      self.end(false, 'heartbeat timeout');
+      self.end('heartbeat timeout');
     }, this.manager.get('heartbeat timeout') * 1000);
 
     this.log.debug('set heartbeat timeout for client', this.id);
@@ -276,7 +279,7 @@ Transport.prototype.setHeartbeatTimeout = function () {
  */
 
 Transport.prototype.clearHeartbeatTimeout = function () {
-  if (this.heartbeatTimeout) {
+  if (this.heartbeatTimeout && this.manager.enabled('heartbeats')) {
     clearTimeout(this.heartbeatTimeout);
     this.heartbeatTimeout = null;
     this.log.debug('cleared heartbeat timeout for client', this.id);
@@ -291,11 +294,12 @@ Transport.prototype.clearHeartbeatTimeout = function () {
  */
 
 Transport.prototype.setHeartbeatInterval = function () {
-  if (!this.heartbeatTimeout) {
+  if (!this.heartbeatInterval && this.manager.enabled('heartbeats')) {
     var self = this;
 
     this.heartbeatInterval = setTimeout(function () {
       self.heartbeat();
+      self.heartbeatInterval = null;
     }, this.manager.get('heartbeat interval') * 1000);
 
     this.log.debug('set heartbeat interval for client', this.id);
@@ -338,25 +342,52 @@ Transport.prototype.heartbeat = function () {
  */
 
 Transport.prototype.onMessage = function (packet) {
+  var current = this.manager.transports[this.id];
+
   if ('heartbeat' == packet.type) {
     this.log.debug('got heartbeat packet');
-    this.store.heartbeat(this.id);
-  } else if ('disconnect' == packet.type && packet.endpoint == '') {
-    this.log.debug('got disconnection packet');
-    this.store.disconnect(this.id, true);
+
+    if (current && current.open) {
+      current.onHeartbeatClear();
+    } else {
+      this.store.publish('heartbeat-clear:' + this.id);
+    }
   } else {
-    this.log.debug('got packet');
+    if ('disconnect' == packet.type && packet.endpoint == '') {
+      this.log.debug('got disconnection packet');
+
+      if (current) {
+        current.onForcedDisconnect();
+      } else {
+        this.store.publish('disconnect-force:' + this.id);
+      }
+
+      return;
+    }
 
     if (packet.id && packet.ack != 'data') {
       this.log.debug('acknowledging packet automatically');
-      this.store.client(this.id).publish(parser.encodePacket({
+
+      var ack = parser.encodePacket({
           type: 'ack'
         , ackId: packet.id
         , endpoint: packet.endpoint || ''
-      }));
+      });
+
+      if (current && current.open) {
+        current.onDispatch(ack);
+      } else {
+        this.manager.onClientDispatch(this.id, ack);
+        this.store.publish('dispatch:' + this.id, ack);
+      }
     }
 
-    this.store.message(this.id, packet);
+    // handle packet locally or publish it
+    if (current) {
+      this.manager.onClientMessage(this.id, packet);
+    } else {
+      this.store.publish('message:' + this.id, packet);
+    }
   }
 };
 
@@ -367,7 +398,7 @@ Transport.prototype.onMessage = function (packet) {
  */
 
 Transport.prototype.clearHeartbeatInterval = function () {
-  if (this.heartbeatInterval) {
+  if (this.heartbeatInterval && this.manager.enabled('heartbeats')) {
     clearTimeout(this.heartbeatInterval);
     this.heartbeatInterval = null;
     this.log.debug('cleared heartbeat interval for client', this.id);
@@ -382,7 +413,7 @@ Transport.prototype.clearHeartbeatInterval = function () {
 
 Transport.prototype.disconnect = function (reason) {
   this.packet({ type: 'disconnect' });
-  this.end(false, reason);
+  this.end(reason);
 
   return this;
 };
@@ -409,8 +440,10 @@ Transport.prototype.close = function () {
 Transport.prototype.onClose = function () {
   if (this.open) {
     this.setCloseTimeout();
-    this.unsubscribe();
+    this.clearHandlers();
     this.open = false;
+    this.manager.onClose(this.id);
+    this.store.publish('close', this.id);
   }
 };
 
@@ -420,43 +453,36 @@ Transport.prototype.onClose = function () {
  * @api private
  */
 
-Transport.prototype.end = function (forced, reason) {
+Transport.prototype.end = function (reason) {
   if (!this.disconnected) {
-    this.log.info('ending socket');
+    this.log.info('transport end');
+
+    var local = this.manager.transports[this.id];
+
     this.close();
     this.clearTimeouts();
-
-    if (!forced)
-      this.store.disconnect(this.id, false, reason);
-
     this.disconnected = true;
+
+    if (local) {
+      this.manager.onClientDisconnect(this.id, reason, true);
+    } else {
+      this.store.publish('disconnect:' + this.id, reason);
+    }
   }
 };
 
-/**
- * Signals that the transport can start flushing buffers.
- *
- * @api public
- */
-
-Transport.prototype.resume = function () {
-  if (!this.disconnected) {
-    this.paused = false;
-    this.setHeartbeatInterval();
-    this.subscribe();
-  }
-
-  return this;
-};
-
 /**
  * Signals that the transport should pause and buffer data.
  *
  * @api public
  */
 
-Transport.prototype.pause = function () {
-  this.paused = true;
+Transport.prototype.discard = function () {
+  this.log.debug('discarding transport');
+  this.discarded = true;
+  this.clearTimeouts();
+  this.clearHandlers();
+
   return this;
 };
 
@@ -476,7 +502,7 @@ Transport.prototype.error = function (reason, advice) {
   });
 
   this.log.warn(reason, advice ? ('client should ' + advice) : '');
-  this.end(false, 'error');
+  this.end('error');
 };
 
 /**
@@ -489,44 +515,6 @@ Transport.prototype.packet = function (obj) {
   return this.write(parser.encodePacket(obj));
 };
 
-/**
- * Subscribe client.
- *
- * @api private
- */
-
-Transport.prototype.subscribe = function () {
-  if (!this.subscribed) {
-    this.log.debug('subscribing', this.id);
-
-    var self = this;
-
-    // subscribe to buffered + normal messages
-    this.store.client(this.id).consume(function (payload, packet) {
-      if (payload) {
-        self.payload(payload);
-      } else {
-        self.write(packet);
-      }
-    });
-
-    this.subscribed = true;
-  }
-};
-
-/**
- * Unsubscribe client.
- *
- * @api private
- */
-
-Transport.prototype.unsubscribe = function () {
-  this.log.info('unsubscribing', this.id);
-
-  this.store.client(this.id).pause();
-  this.subscribed = false;
-};
-
 /**
  * Writes a volatile message.
  *

lib/transports/flashsocket.js

@@ -23,8 +23,8 @@ exports = module.exports = FlashSocket;
  * @api public
  */
  
-function FlashSocket () {
-  WebSocket.apply(this, arguments);
+function FlashSocket (mng, data, req) {
+  WebSocket.call(this, mng, data, req);
 }
 
 /**
@@ -33,6 +33,14 @@ function FlashSocket () {
 
 FlashSocket.prototype.__proto__ = WebSocket.prototype;
 
+/**
+ * Transport name
+ *
+ * @api public
+ */
+
+FlashSocket.prototype.name = 'flashsocket';
+
 /**
  * Listens for new configuration changes of the Manager
  * this way we can enable and disable the flash server.

lib/transports/htmlfile.js

@@ -23,8 +23,8 @@ exports = module.exports = HTMLFile;
  * @api public
  */
 
-function HTMLFile (mng, data) {
-  HTTPTransport.call(this, mng, data);
+function HTMLFile (mng, data, req) {
+  HTTPTransport.call(this, mng, data, req);
 };
 
 /**
@@ -33,6 +33,14 @@ function HTMLFile (mng, data) {
 
 HTMLFile.prototype.__proto__ = HTTPTransport.prototype;
 
+/**
+ * Transport name
+ *
+ * @api public
+ */
+
+HTMLFile.prototype.name = 'htmlfile';
+
 /**
  * Handles the request.
  *
@@ -70,5 +78,5 @@ HTMLFile.prototype.write = function (data) {
     this.drained = true;
   }
 
-  this.log.debug('htmlfile writing', data);
+  this.log.debug(this.name + ' writing', data);
 };

lib/transports/http-polling.js

@@ -23,8 +23,8 @@ exports = module.exports = HTTPPolling;
  * @api public.
  */
 
-function HTTPPolling (mng, data) {
-  HTTPTransport.call(this, mng, data);
+function HTTPPolling (mng, data, req) {
+  HTTPTransport.call(this, mng, data, req);
 };
 
 /**
@@ -35,6 +35,14 @@ function HTTPPolling (mng, data) {
 
 HTTPPolling.prototype.__proto__ = HTTPTransport.prototype;
 
+/**
+ * Transport name
+ *
+ * @api public
+ */
+
+HTTPPolling.prototype.name = 'httppolling';
+
 /**
  * Removes heartbeat timeouts for polling.
  */
@@ -57,7 +65,7 @@ HTTPPolling.prototype.handleRequest = function (req) {
 
     this.pollTimeout = setTimeout(function () {
       self.packet({ type: 'noop' });
-      self.log.debug('polling closed due to exceeded duration');
+      self.log.debug(self.name + ' closed due to exceeded duration');
     }, this.manager.get('polling duration') * 1000);
 
     this.log.debug('setting poll timeout');

lib/transports/http.js

@@ -25,8 +25,8 @@ exports = module.exports = HTTPTransport;
  * @api public
  */
 
-function HTTPTransport (mng, data) {
-  Transport.call(this, mng, data);
+function HTTPTransport (mng, data, req) {
+  Transport.call(this, mng, data, req);
 };
 
 /**
@@ -54,6 +54,9 @@ HTTPTransport.prototype.handleRequest = function (req) {
     });
 
     req.on('end', function () {
+      res.writeHead(200, headers);
+      res.end('1');
+      
       self.onData(self.postEncoded ? qs.parse(buffer).d : buffer);
     });
 
@@ -65,9 +68,6 @@ HTTPTransport.prototype.handleRequest = function (req) {
         headers['Access-Control-Allow-Credentials'] = 'true';
       }
     }
-
-    res.writeHead(200, headers);
-    res.end('1');
   } else {
     this.response = req.res;
 
@@ -85,7 +85,7 @@ HTTPTransport.prototype.onData = function (data) {
   var messages = parser.decodePayload(data);
 
   for (var i = 0, l = messages.length; i < l; i++) {
-    this.log.debug('xhr received data packet', data);
+    this.log.debug(this.name + ' received data packet', data);
     this.onMessage(messages[i]);
   }
 };

lib/transports/jsonp-polling.js

@@ -23,8 +23,8 @@ exports = module.exports = JSONPPolling;
  * @api public
  */
 
-function JSONPPolling (mng, data) {
-  HTTPPolling.call(this, mng, data);
+function JSONPPolling (mng, data, req) {
+  HTTPPolling.call(this, mng, data, req);
 
   this.head = 'io.j[0](';
   this.foot = ');';
@@ -40,6 +40,14 @@ function JSONPPolling (mng, data) {
 
 JSONPPolling.prototype.__proto__ = HTTPPolling.prototype;
 
+/**
+ * Transport name
+ *
+ * @api public
+ */
+
+JSONPPolling.prototype.name = 'jsonppolling';
+
 /**
  * Make sure POST are decoded.
  */
@@ -62,8 +70,9 @@ JSONPPolling.prototype.doWrite = function (data) {
       'Content-Type': 'text/javascript; charset=UTF-8'
     , 'Content-Length': Buffer.byteLength(data)
     , 'Connection': 'Keep-Alive'
+    , 'X-XSS-Protection': '0'
   });
 
   this.response.write(data);
-  this.log.debug('json-p writing', data);
+  this.log.debug(this.name + ' writing', data);
 };

lib/transports/websocket.js

@@ -27,13 +27,13 @@ exports = module.exports = WebSocket;
  * @api public
  */
 
-function WebSocket (mng, data) {
+function WebSocket (mng, data, req) {
   // parser
   var self = this;
 
   this.parser = new Parser();
   this.parser.on('data', function (packet) {
-    self.log.debug('websocket received data packet', packet);
+    self.log.debug(self.name + ' received data packet', packet);
     self.onMessage(parser.decodePacket(packet));
   });
   this.parser.on('close', function () {
@@ -43,7 +43,7 @@ function WebSocket (mng, data) {
     self.end();
   });
 
-  Transport.call(this, mng, data);
+  Transport.call(this, mng, data, req);
 };
 
 /**
@@ -52,6 +52,14 @@ function WebSocket (mng, data) {
 
 WebSocket.prototype.__proto__ = Transport.prototype;
 
+/**
+ * Transport name
+ *
+ * @api public
+ */
+
+WebSocket.prototype.name = 'websocket';
+
 /**
  * Called when the socket connects.
  *
@@ -67,7 +75,7 @@ WebSocket.prototype.onSocketConnect = function () {
   this.buffered = [];
 
   if (this.req.headers.upgrade !== 'WebSocket') {
-    this.log.warn('WebSocket connection invalid');
+    this.log.warn(this.name + ' connection invalid');
     this.end();
     return;
   }
@@ -180,7 +188,7 @@ WebSocket.prototype.write = function (data) {
       this.end();
     }
 
-    this.log.debug('websocket writing', data);
+    this.log.debug(this.name + ' writing', data);
   }
 };
 
@@ -194,7 +202,7 @@ WebSocket.prototype.flush = function () {
   this.buffer = false;
 
   for (var i = 0, l = this.buffered.length; i < l; i++) {
-    this.write(this.buffered.splice(0, 1));
+    this.write(this.buffered.splice(0, 1)[0]);
   }
 };
 
@@ -217,7 +225,7 @@ WebSocket.prototype.proveReception = function (headers) {
         , spaces = k.replace(/[^ ]/g, '').length;
 
       if (spaces === 0 || n % spaces !== 0){
-        self.log.warn('Invalid WebSocket key: "' + k + '".');
+        self.log.warn('Invalid ' + self.name + ' key: "' + k + '".');
         self.end();
         return false;
       }

lib/transports/xhr-polling.js

@@ -23,8 +23,8 @@ exports = module.exports = XHRPolling;
  * @api public
  */
 
-function XHRPolling (mng, data) {
-  HTTPPolling.call(this, mng, data);
+function XHRPolling (mng, data, req) {
+  HTTPPolling.call(this, mng, data, req);
 };
 
 /**
@@ -33,6 +33,14 @@ function XHRPolling (mng, data) {
 
 XHRPolling.prototype.__proto__ = HTTPPolling.prototype;
 
+/**
+ * Transport name
+ *
+ * @api public
+ */
+
+XHRPolling.prototype.name = 'xhr-polling';
+
 /**
  * Frames data prior to write.
  *
@@ -60,5 +68,5 @@ XHRPolling.prototype.doWrite = function (data) {
 
   this.response.writeHead(200, headers);
   this.response.write(data);
-  this.log.debug('xhr-polling writing', data);
+  this.log.debug(this.name + ' writing', data);
 };

package.json

@@ -1,7 +1,7 @@
 {
     "name": "socket.io"
-  , "version": "0.7.2"
-  , "description": "Realtime apps made cross-browser & easy with a WebSocket-like API"
+  , "version": "0.7.8"
+  , "description": "Real-time apps made cross-browser & easy with a WebSocket-like API"
   , "homepage": "http://socket.io"
   , "keywords": ["websocket", "socket", "realtime", "socket.io", "comet", "ajax"]
   , "author": "Guillermo Rauch <guillermo@learnboost.com>"
@@ -15,12 +15,14 @@
       , "url": "https://github.com/LearnBoost/Socket.IO-node.git"
     }
   , "dependencies": {
-        "socket.io-client": "0.7.2"
-      , "policyfile": ">= 0.0.3"
+        "socket.io-client": "0.7.5"
+      , "policyfile": "0.0.4"
+      , "redis": "0.6.6"
     }
   , "devDependencies": {
         "expresso": "0.7.7"
       , "should": "0.0.4"
+      , "assertvanish": "0.0.3-1"
     }
   , "main": "index"
   , "engines": { "node": ">= 0.4.0" }

test/common.js

@@ -12,7 +12,8 @@
 var io = require('socket.io')
   , parser = io.parser
   , http = require('http')
-  , https = require('https');
+  , https = require('https')
+  , WebSocket = require('../support/node-websocket-client/lib/websocket').WebSocket;
 
 /**
  * Exports.
@@ -178,6 +179,66 @@ client = function (port) {
  */
 
 create = function (cl) {
-  console.error('');
-  return io.listen(cl.port);
+  console.log('');
+  var manager = io.listen(cl.port);
+  manager.set('client store expiration', 0);
+  return manager;
+};
+
+/**
+ * WebSocket socket.io client.
+ *
+ * @api private
+ */
+
+function WSClient (port, sid) {
+  this.sid = sid;
+  this.port = port;
+
+  WebSocket.call(
+      this
+    , 'ws://localhost:' + port + '/socket.io/' 
+        + io.protocol + '/websocket/' + sid
+  );
+};
+
+/**
+ * Inherits from WebSocket.
+ */
+
+WSClient.prototype.__proto__ = WebSocket.prototype;
+
+/**
+ * Overrides message event emission.
+ *
+ * @api private
+ */
+
+WSClient.prototype.emit = function (name) {
+  var args = arguments;
+
+  if (name == 'message' || name == 'data') {
+    args[1] = parser.decodePacket(args[1].toString());
+  }
+
+  return WebSocket.prototype.emit.apply(this, arguments);
+};
+
+/**
+ * Writes a packet
+ */
+
+WSClient.prototype.packet = function (pack) {
+  this.write(parser.encodePacket(pack));
+  return this;
+};
+
+/**
+ * Creates a websocket client.
+ *
+ * @api public
+ */
+
+websocket = function (cl, sid) {
+  return new WSClient(cl.port, sid);
 };

test/leaks/socket.leaktest.js

@@ -0,0 +1,54 @@
+/*!
+ * socket.io-node
+ * Copyright(c) 2011 LearnBoost <dev@learnboost.com>
+ * MIT Licensed
+ */
+
+/**
+ * Test dependencies.
+ */
+
+require.paths.unshift(__dirname + '/../../lib');
+
+var assertvanish = require('assertvanish')
+  , common = require('../common')
+  , ports = 15800;
+
+function resultCallback (leaks, leakedSocket) {
+  if (leaks) {
+    console.error('Leak detected');
+    process.exit(1);
+  } else {
+    console.error('No leaks');
+    process.exit(0);
+  }
+};
+
+/**
+ * Test.
+ */
+
+var cl = client(++ports);
+var io = create(cl);
+
+io.sockets.on('connection', function (socket) {
+  console.log('connected');
+  
+  socket.on('disconnect', function() {
+    console.log("client gone");
+    setTimeout(gc, 1000);
+    assertvanish(socket, 2000, {silent: true, callback: resultCallback});
+  });
+});
+
+setTimeout(function() {
+  cl.handshake(function (sid) {
+    var ws = websocket(cl, sid);
+    ws.on('open', function () {
+      console.log('open!');
+      setTimeout(function() {
+        ws.close();
+      }, 500);
+    });
+  });
+}, 100);

test/manager.test.js

@@ -1,4 +1,3 @@
-
 /*!
  * socket.io-node
  * Copyright(c) 2011 LearnBoost <dev@learnboost.com>
@@ -21,6 +20,22 @@ var sio = require('socket.io')
 module.exports = {
 
   'test setting and getting a configuration flag': function (done) {
+    var port = ++ports
+      , io = sio.listen(http.createServer());
+
+    io.set('a', 'b');
+    io.get('a').should.eql('b');
+
+    var port = ++ports
+      , io = sio.listen(http.createServer());
+
+    io.configure(function () {
+      io.set('a', 'b');
+      io.enable('tobi');
+    });
+
+    io.get('a').should.eql('b');
+
     done();
   },
 
@@ -37,11 +52,9 @@ module.exports = {
       , io = sio.listen(http.createServer());
 
     io.configure(function () {
-      io.set('a', 'b');
       io.enable('tobi');
     });
 
-    io.get('a').should.eql('b');
     io.enabled('tobi').should.be.true;
 
     done();
@@ -237,11 +250,37 @@ module.exports = {
     });
   },
 
-  'test that client minification works': function (done) {
+  'test that the cached client sends a 304 header': function (done) {
     var port = ++ports
       , io = sio.listen(port)
       , cl = client(port);
 
+     io.configure(function () {
+      io.enable('browser client etag');
+    });
+
+    cl.get('/socket.io/socket.io.js', function (res, data) {
+      cl.get('/socket.io/socket.io.js', {headers:{'if-none-match':res.headers.etag}}, function (res, data) {
+        res.statusCode.should.eql(304);
+
+        cl.end();
+        io.server.close();
+        done();
+      });
+    });
+  },
+
+  'test that client minification works': function (done) {
+    // server 1
+    var port = ++ports
+      , io = sio.listen(port)
+      , cl = client(port);
+
+    // server 2
+    var port = ++ports
+      , io2 = sio.listen(port)
+      , cl2 = client(port);
+
     io.configure(function () {
       io.enable('browser client minification');
     });
@@ -252,11 +291,6 @@ module.exports = {
       cl.end();
       io.server.close();
 
-      // start a new server with minification enabled and compare lengths
-      var port = ++ports
-        , io2 = sio.listen(port)
-        , cl2 = client(port);
-
       cl2.get('/socket.io/socket.io.js', function (res, data) {
         res.headers['content-type'].should.eql('application/javascript');
         res.headers['content-length'].should.match(/([0-9]+)/);
@@ -513,6 +547,100 @@ module.exports = {
       io.server.close();
       done();
     });
-  }
+  },
 
+  'test disabling heartbeats': function (done) {
+    var port = ++ports
+      , io = sio.listen(port)
+      , cl = client(port)
+      , messages = 0
+      , beat = false
+      , ws;
+
+    io.configure(function () {
+      io.disable('heartbeats');
+      io.set('heartbeat interval', .05);
+      io.set('heartbeat timeout', .05);
+      io.set('close timeout', .05);
+    });
+
+    io.sockets.on('connection', function (socket) {
+      setTimeout(function () {
+        socket.disconnect();
+      }, io.get('heartbeat timeout') * 1000 + 100);
+
+      socket.on('disconnect', function (reason) {
+        beat.should.be.false;
+
+        cl.end();
+        ws.finishClose();
+        io.server.close();
+        done();
+      });
+    });
+
+    cl.get('/socket.io/{protocol}/', function (res, data) {
+      res.statusCode.should.eql(200);
+      data.should.match(/([^:]+)::[\.0-9]+:(.*)/);
+
+      cl.handshake(function (sid) {
+        ws = websocket(cl, sid);
+        ws.on('message', function (packet) {
+          if (++messages == 1) {
+            packet.type.should.eql('connect');
+          } else if (packet.type == 'heartbeat'){
+            beat = true;
+          }
+        });
+      });
+    });
+  },
+
+  'no duplicate room members': function (done) {
+    var port = ++ports
+      , io = sio.listen(port);
+
+    Object.keys(io.rooms).length.should.equal(0);
+
+    io.onJoin(123, 'foo');
+    io.rooms.foo.length.should.equal(1);
+
+    io.onJoin(123, 'foo');
+    io.rooms.foo.length.should.equal(1);
+
+    io.onJoin(124, 'foo');
+    io.rooms.foo.length.should.equal(2);
+
+    io.onJoin(124, 'foo');
+    io.rooms.foo.length.should.equal(2);
+
+    io.onJoin(123, 'bar');
+    io.rooms.foo.length.should.equal(2);
+    io.rooms.bar.length.should.equal(1);
+
+    io.onJoin(123, 'bar');
+    io.rooms.foo.length.should.equal(2);
+    io.rooms.bar.length.should.equal(1);
+
+    io.onJoin(124, 'bar');
+    io.rooms.foo.length.should.equal(2);
+    io.rooms.bar.length.should.equal(2);
+
+    io.onJoin(124, 'bar');
+    io.rooms.foo.length.should.equal(2);
+    io.rooms.bar.length.should.equal(2);
+
+    io.server.close();
+    done();
+  },
+
+  'test passing options directly to the Manager through listen': function (done) {
+    var port = ++ports
+      , io = sio.listen(port, { resource: '/my resource', custom: 'opt' });
+
+    io.get('resource').should.equal('/my resource');
+    io.get('custom').should.equal('opt');
+    io.server.close();
+    done();
+  }
 };

test/namespace.test.js

@@ -0,0 +1,247 @@
+
+/*!
+ * socket.io-node
+ * Copyright(c) 2011 LearnBoost <dev@learnboost.com>
+ * MIT Licensed
+ */
+
+/**
+ * Test dependencies.
+ */
+
+var sio = require('socket.io')
+  , should = require('./common')
+  , ports = 15700;
+
+/**
+ * Test.
+ */
+
+module.exports = {
+  'namespace pass no authentication': function (done) {
+    var cl = client(++ports)
+      , io = create(cl)
+      , ws;
+
+    io.of('/a')
+      .on('connection', function (socket) {
+        cl.end();
+        ws.finishClose();
+        io.server.close()
+        done();
+      });
+
+    cl.handshake(function (sid) {
+      ws = websocket(cl, sid);
+      ws.on('open', function () {
+        ws.packet({
+            type: 'connect'
+          , endpoint: '/a'
+        });
+      })
+    });
+  },
+
+  'namespace pass authentication': function (done) {
+    var cl = client(++ports)
+      , io = create(cl)
+      , ws;
+
+    io.of('/a')
+      .authorization(function (data, fn) {
+        fn(null, true);
+      })
+      .on('connection', function (socket) {
+        cl.end();
+        ws.finishClose();
+        io.server.close()
+        done();
+      });
+
+    cl.handshake(function (sid) {
+      ws = websocket(cl, sid);
+      ws.on('open', function () {
+        ws.packet({
+            type: 'connect'
+          , endpoint: '/a'
+        });
+      })
+    });
+  },
+
+  'namespace authentication handshake data': function (done) {
+    var cl = client(++ports)
+      , io = create(cl)
+      , ws;
+
+    io.of('/a')
+      .authorization(function (data, fn) {
+        data.foo = 'bar';
+        fn(null, true);
+      })
+      .on('connection', function (socket) {
+        (!!socket.handshake.address.address).should.be.true;
+        (!!socket.handshake.address.port).should.be.true;
+        socket.handshake.headers.host.should.equal('localhost');
+        socket.handshake.headers.connection.should.equal('keep-alive');
+        socket.handshake.time.should.match(/GMT/);
+        socket.handshake.foo.should.equal('bar');
+
+        cl.end();
+        ws.finishClose();
+        io.server.close()
+        done();
+      });
+
+    cl.handshake(function (sid) {
+      ws = websocket(cl, sid);
+      ws.on('open', function () {
+        ws.packet({
+            type: 'connect'
+          , endpoint: '/a'
+        });
+      })
+    });
+  },
+
+  'namespace fail authentication': function (done) {
+    var cl = client(++ports)
+      , io = create(cl)
+      , calls = 0
+      , ws;
+
+    io.of('/a')
+      .authorization(function (data, fn) {
+        fn(null, false);
+      })
+      .on('connection', function (socket) {
+        throw new Error('Should not be called');
+      });
+
+    cl.handshake(function (sid) {
+      ws = websocket(cl, sid);
+      ws.on('open', function () {
+        ws.packet({
+            type: 'connect'
+          , endpoint: '/a'
+        });
+      });
+
+      ws.on('message', function (data) {
+        if (data.endpoint == '/a') {
+          data.type.should.eql('error');
+          data.reason.should.eql('unauthorized')
+          
+          cl.end();
+          ws.finishClose();
+          io.server.close()
+          done();
+        }
+      })
+    });
+  },
+
+  'broadcasting sends and emits on a namespace': function (done) {
+    var cl = client(++ports)
+      , io = create(cl)
+      , calls = 0
+      , connect = 0
+      , message = 0
+      , events = 0
+      , expected = 5
+      , ws1
+      , ws2;
+
+    io.of('a')
+      .on('connection', function (socket){
+        socket.broadcast.emit('b', 'test');
+        socket.broadcast.json.emit('json', {foo:'bar'});
+        socket.broadcast.send('foo');
+      });
+
+    function finish () {
+      connect.should.equal(2);
+      message.should.equal(1);
+      events.should.equal(2);
+
+      cl.end();
+      ws1.finishClose();
+      ws2.finishClose();
+      io.server.close();
+      done();
+    }
+
+    cl.handshake(function (sid) {
+     ws1 = websocket(cl, sid);
+
+      ws1.on('open', function() {
+        ws1.packet({
+            type: 'connect'
+          , endpoint: 'a'
+        });
+      });
+
+      ws1.on('message', function (data) {
+        if (data.type === 'connect') {
+          ++connect;
+          if (++calls === expected) finish();
+        }
+
+        if (data.type === 'message') {
+          ++message;
+          if (++calls === expected) finish();
+        }
+
+        if (data.type === 'event') {
+          if (data.name === 'b' || data.name === 'json') ++events;
+          if (++calls === expected) finish();
+        }
+      });
+
+      cl.handshake(function (sid) {
+        ws2 = websocket(cl, sid);
+
+        ws2.on('open', function () {
+          ws2.packet({
+              type: 'connect'
+            , endpoint: 'a'
+          });
+        });
+      })
+    })
+  },
+
+  'joining rooms inside a namespace': function (done) {
+    var cl = client(++ports)
+      , io = create(cl)
+      , calls = 0
+      , ws;
+
+    io.of('/foo').on('connection', function (socket) {
+      socket.join('foo.bar');
+      this.in('foo.bar').emit('baz', 'pewpew');
+    });
+
+    cl.handshake(function (sid) {
+      ws = websocket(cl, sid);
+
+      ws.on('open', function (){
+         ws.packet({
+            type: 'connect'
+          , endpoint: '/foo'
+        });
+      });
+
+      ws.on('message', function (data) {
+        if (data.type === 'event') {
+          data.name.should.equal('baz');
+
+          cl.end();
+          ws.finishClose();
+          io.server.close();
+          done();
+        }
+      });
+    })
+  }
+};

test/stores.memory.test.js

@@ -0,0 +1,190 @@
+
+/**
+ * Test dependencies
+ *
+ * @api private
+ */
+
+var sio = require('socket.io')
+  , should = require('should')
+  , MemoryStore = sio.MemoryStore;
+
+/**
+ * Test.
+ */
+
+module.exports = {
+
+  'test storing data for a client': function (done) {
+    var store = new MemoryStore
+      , client = store.client('test');
+
+    client.id.should.equal('test');
+
+    client.set('a', 'b', function (err) {
+      should.strictEqual(err, null);
+
+      client.get('a', function (err, val) {
+        should.strictEqual(err, null);
+        val.should.eql('b');
+
+        client.has('a', function (err, has) {
+          should.strictEqual(err, null);
+          has.should.be.true;
+
+          client.has('b', function (err, has) {
+            should.strictEqual(err, null);
+            has.should.be.false;
+
+            client.del('a', function (err) {
+              should.strictEqual(err, null);
+
+              client.has('a', function (err, has) {
+                should.strictEqual(err, null);
+                has.should.be.false;
+
+                client.set('b', 'c', function (err) {
+                  should.strictEqual(err, null);
+                  
+                  client.set('c', 'd', function (err) {
+                    should.strictEqual(err, null);
+
+                    client.get('b', function (err, val) {
+                      should.strictEqual(err, null);
+                      val.should.equal('c');
+
+                      client.get('c', function (err, val) {
+                        should.strictEqual(err, null);
+                        val.should.equal('d');
+
+                        store.destroy();
+                        done();
+                      });
+                    });
+                  });
+                });
+              });
+            });
+          });
+        });
+      });
+    });
+  },
+
+  'test cleaning up clients data': function (done) {
+    var rand1 = Math.abs(Math.random() * Date.now() | 0)
+      , rand2 = Math.abs(Math.random() * Date.now() | 0);
+
+    var store = new MemoryStore()
+      , client1 = store.client(rand1)
+      , client2 = store.client(rand2);
+
+    client1.set('a', 'b', function (err) {
+      should.strictEqual(err, null);
+
+      client2.set('c', 'd', function (err) {
+        should.strictEqual(err, null);
+
+        client1.has('a', function (err, val) {
+          should.strictEqual(err, null);
+          val.should.be.true;
+
+          client2.has('c', function (err, val) {
+            should.strictEqual(err, null);
+            val.should.be.true;
+
+            store.destroy();
+
+            var newstore = new MemoryStore()
+              , newclient1 = newstore.client(rand1)
+              , newclient2 = newstore.client(rand2);
+
+            newclient1.has('a', function (err, val) {
+              should.strictEqual(err, null);
+              val.should.be.false;
+
+              newclient2.has('c', function (err, val) {
+                should.strictEqual(err, null);
+                val.should.be.false;
+
+                newstore.destroy();
+                done();
+              });
+            });
+          });
+        });
+      });
+    });
+  },
+
+  'test cleaning up a particular client': function (done) {
+    var rand1 = Math.abs(Math.random() * Date.now() | 0)
+      , rand2 = Math.abs(Math.random() * Date.now() | 0);
+
+    var store = new MemoryStore()
+      , client1 = store.client(rand1)
+      , client2 = store.client(rand2);
+
+    client1.set('a', 'b', function (err) {
+      should.strictEqual(err, null);
+
+      client2.set('c', 'd', function (err) {
+        should.strictEqual(err, null);
+
+        client1.has('a', function (err, val) {
+          should.strictEqual(err, null);
+          val.should.be.true;
+
+          client2.has('c', function (err, val) {
+            should.strictEqual(err, null);
+            val.should.be.true;
+
+            store.clients.should.have.property(rand1);
+            store.clients.should.have.property(rand2);
+            store.destroyClient(rand1);
+
+            store.clients.should.not.have.property(rand1);
+            store.clients.should.have.property(rand2);
+
+            client1.has('a', function (err, val) {
+              should.strictEqual(err, null);
+              val.should.equal(false);
+
+              store.destroy();
+              done();
+            });
+          });
+        });
+      });
+    });
+  },
+
+  'test destroy expiration': function (done) {
+    var store = new MemoryStore()
+      , id = Math.abs(Math.random() * Date.now() | 0)
+      , client = store.client(id);
+
+    client.set('a', 'b', function (err) {
+      should.strictEqual(err, null);
+      store.destroyClient(id, 1);
+
+      setTimeout(function () {
+        client.get('a', function (err, val) {
+          should.strictEqual(err, null);
+          val.should.equal('b');
+        });
+      }, 500);
+
+      setTimeout(function () {
+        client.get('a', function (err, val) {
+          should.strictEqual(err, null);
+          should.strictEqual(val, null);
+
+          store.destroy();
+          done();
+        });
+      }, 1900);
+    });
+  }
+
+};

test/stores.redis.test.js

@@ -0,0 +1,240 @@
+
+/**
+ * Test dependencies
+ *
+ * @api private
+ */
+
+var sio = require('socket.io')
+  , redis = require('redis')
+  , should = require('should')
+  , RedisStore = sio.RedisStore;
+
+/**
+ * Test.
+ */
+
+module.exports = {
+
+  'test publishing doesnt get caught by the own store subscriber': function (done) {
+    var a = new RedisStore
+      , b = new RedisStore;
+
+    a.subscribe('woot', function (arg) {
+      arg.should.equal('bb');
+      a.destroy();
+      b.destroy();
+      done();
+    }, function () {
+      a.publish('woot', 'aa');
+      b.publish('woot', 'bb');
+    });
+  },
+
+  'test publishing to multiple subscribers': function (done) {
+    var a = new RedisStore
+      , b = new RedisStore
+      , c = new RedisStore
+      , subscriptions = 3
+      , messages = 2;
+
+    a.subscribe('tobi', function () {
+      throw new Error('Shouldnt publish to itself');
+    }, publish);
+
+    function subscription (arg1, arg2, arg3) {
+      arg1.should.equal(1);
+      arg2.should.equal(2);
+      arg3.should.equal(3);
+      --messages || finish();
+    }
+
+    b.subscribe('tobi', subscription, publish);
+    c.subscribe('tobi', subscription, publish);
+
+    function publish () {
+      --subscriptions || a.publish('tobi', 1, 2, 3);
+    }
+
+    function finish () {
+      a.destroy();
+      b.destroy();
+      c.destroy();
+      done();
+    }
+  },
+
+  'test storing data for a client': function (done) {
+    var store = new RedisStore
+      , rand = 'test-' + Date.now()
+      , client = store.client(rand);
+
+    client.id.should.equal(rand);
+
+    client.set('a', 'b', function (err) {
+      should.strictEqual(err, null);
+
+      client.get('a', function (err, val) {
+        should.strictEqual(err, null);
+        val.should.equal('b');
+
+        client.has('a', function (err, has) {
+          should.strictEqual(err, null);
+          has.should.be.true;
+
+          client.has('b', function (err, has) {
+            should.strictEqual(err, null);
+            has.should.be.false;
+
+            client.del('a', function (err) {
+              should.strictEqual(err, null);
+
+              client.has('a', function (err, has) {
+                should.strictEqual(err, null);
+                has.should.be.false;
+
+                client.set('b', 'c', function (err) {
+                  should.strictEqual(err, null);
+                  
+                  client.set('c', 'd', function (err) {
+                    should.strictEqual(err, null);
+
+                    client.get('b', function (err, val) {
+                      should.strictEqual(err, null);
+                      val.should.equal('c');
+
+                      client.get('c', function (err, val) {
+                        should.strictEqual(err, null);
+                        val.should.equal('d');
+
+                        store.destroy();
+                        done();
+                      });
+                    });
+                  });
+                });
+              });
+            });
+          });
+        });
+      });
+    });
+  },
+
+  'test cleaning up clients data': function (done) {
+    var rand1 = Math.abs(Math.random() * Date.now() | 0)
+      , rand2 = Math.abs(Math.random() * Date.now() | 0);
+
+    var store = new RedisStore()
+      , client1 = store.client(rand1)
+      , client2 = store.client(rand2);
+
+    client1.set('a', 'b', function (err) {
+      should.strictEqual(err, null);
+
+      client2.set('c', 'd', function (err) {
+        should.strictEqual(err, null);
+
+        client1.has('a', function (err, val) {
+          should.strictEqual(err, null);
+          val.should.be.true;
+
+          client2.has('c', function (err, val) {
+            should.strictEqual(err, null);
+            val.should.be.true;
+
+            store.destroy();
+
+            var newstore = new RedisStore()
+              , newclient1 = newstore.client(rand1)
+              , newclient2 = newstore.client(rand2);
+
+            newclient1.has('a', function (err, val) {
+              should.strictEqual(err, null);
+              val.should.be.false;
+
+              newclient2.has('c', function (err, val) {
+                should.strictEqual(err, null);
+                val.should.be.false;
+
+                newstore.destroy();
+                done();
+              });
+            });
+          });
+        });
+      });
+    });
+  },
+
+  'test cleaning up a particular client': function (done) {
+    var rand1 = Math.abs(Math.random() * Date.now() | 0)
+      , rand2 = Math.abs(Math.random() * Date.now() | 0);
+
+    var store = new RedisStore()
+      , client1 = store.client(rand1)
+      , client2 = store.client(rand2);
+
+    client1.set('a', 'b', function (err) {
+      should.strictEqual(err, null);
+
+      client2.set('c', 'd', function (err) {
+        should.strictEqual(err, null);
+
+        client1.has('a', function (err, val) {
+          should.strictEqual(err, null);
+          val.should.be.true;
+
+          client2.has('c', function (err, val) {
+            should.strictEqual(err, null);
+            val.should.be.true;
+
+            store.clients.should.have.property(rand1);
+            store.clients.should.have.property(rand2);
+            store.destroyClient(rand1);
+
+            store.clients.should.not.have.property(rand1);
+            store.clients.should.have.property(rand2);
+
+            client1.has('a', function (err, val) {
+              should.strictEqual(err, null);
+              val.should.equal(false);
+
+              store.destroy();
+              done();
+            });
+          });
+        });
+      });
+    });
+  },
+
+  'test destroy expiration': function (done) {
+    var store = new RedisStore()
+      , id = Math.abs(Math.random() * Date.now() | 0)
+      , client = store.client(id);
+
+    client.set('a', 'b', function (err) {
+      should.strictEqual(err, null);
+      store.destroyClient(id, 1);
+
+      setTimeout(function () {
+        client.get('a', function (err, val) {
+          should.strictEqual(err, null);
+          val.should.equal('b');
+        });
+      }, 500);
+
+      setTimeout(function () {
+        client.get('a', function (err, val) {
+          should.strictEqual(err, null);
+          should.strictEqual(val, null);
+
+          store.destroy();
+          done();
+        });
+      }, 2000);
+    });
+  }
+
+};

test/transports.jsonp-polling.test.js

@@ -177,7 +177,6 @@ module.exports = {
         --total || finish();
       });
 
-      // we rely on a small poll duration to close this request quickly
       cl.get('/socket.io/{protocol}/jsonp-polling/' + sid, function (res, msgs) {
         res.statusCode.should.eql(200);
         msgs.should.have.length(1);
@@ -244,7 +243,6 @@ module.exports = {
         msgs.should.have.length(1);
         msgs[0].type.should.eql('connect');
 
-        // here we close the request instead of relying on a small poll timeout
         setTimeout(function () {
           cl.end();
         }, 10);

test/transports.websocket.test.js

@@ -11,74 +11,8 @@
 
 var sio = require('socket.io')
   , should = require('./common')
-  , HTTPClient = should.HTTPClient
-  , WebSocket = require('../support/node-websocket-client/lib/websocket').WebSocket
   , parser = sio.parser
-  , ports = 15400;
-
-/**
- * Exports WSClient.
- */
-
-module.exports = exports = WSClient;
-
-/**
- * WebSocket socket.io client.
- *
- * @api private
- */
-
-function WSClient (port, sid) {
-  this.sid = sid;
-  this.port = port;
-
-  WebSocket.call(
-      this
-    , 'ws://localhost:' + port + '/socket.io/' 
-        + sio.protocol + '/websocket/' + sid
-  );
-};
-
-/**
- * Inherits from WebSocket.
- */
-
-WSClient.prototype.__proto__ = WebSocket.prototype;
-
-/**
- * Overrides message event emission.
- *
- * @api private
- */
-
-WSClient.prototype.emit = function (name) {
-  var args = arguments;
-
-  if (name == 'message' || name == 'data') {
-    args[1] = parser.decodePacket(args[1].toString());
-  }
-
-  return WebSocket.prototype.emit.apply(this, arguments);
-};
-
-/**
- * Writes a packet
- */
-
-WSClient.prototype.packet = function (pack) {
-  this.write(parser.encodePacket(pack));
-  return this;
-};
-
-/**
- * Creates a websocket client.
- *
- * @api public
- */
-
-function websocket (cl, sid) {
-  return new WSClient(cl.port, sid);
-};
+  , ports = 15800;
 
 /**
  * Tests.
@@ -89,7 +23,8 @@ module.exports = {
   'test that not responding to a heartbeat drops client': function (done) {
     var cl = client(++ports)
       , io = create(cl)
-      , messages = 0;
+      , messages = 0
+      , ws;
 
     io.configure(function () {
       io.set('heartbeat interval', .05);
@@ -103,13 +38,14 @@ module.exports = {
         reason.should.eql('heartbeat timeout');
 
         cl.end();
+        ws.finishClose();
         io.server.close();
         done();
       });
     });
 
     cl.handshake(function (sid) {
-      var ws = websocket(cl, sid);
+      ws = websocket(cl, sid);
       ws.on('message', function (packet) {
         if (++messages == 1) {
           packet.type.should.eql('connect');
@@ -125,7 +61,8 @@ module.exports = {
     var cl = client(++ports)
       , io = create(cl)
       , messages = 0
-      , heartbeats = 0;
+      , heartbeats = 0
+      , ws;
 
     io.configure(function () {
       io.set('heartbeat interval', .05);
@@ -139,13 +76,14 @@ module.exports = {
         reason.should.eql('heartbeat timeout');
 
         cl.end();
+        ws.finishClose();
         io.server.close();
         done();
       });
     });
 
     cl.handshake(function (sid) {
-      var ws = websocket(cl, sid);
+      ws = websocket(cl, sid);
       ws.on('message', function (packet) {
         if (++messages == 1) {
           packet.type.should.eql('connect');
@@ -644,16 +582,15 @@ module.exports = {
       connections++;
 
       if (connections != 3) {
-        socket.join('woot', function () {
-          joins++;
+        socket.join('woot');
+        joins++;
 
-          if (joins == 2) {
-            setTimeout(function () {
-              connections.should.eql(3);
-              io.sockets.in('woot').send('hahaha');
-            }, 20);
-          }
-        });
+        if (joins == 2) {
+          setTimeout(function () {
+            connections.should.eql(3);
+            io.sockets.in('woot').send('hahaha');
+          }, 20);
+        }
       }
 
       socket.on('disconnect', function () {
@@ -756,16 +693,15 @@ module.exports = {
       connections++;
 
       if (connections != 3) {
-        socket.join('woot', function () {
-          joins++;
+        socket.join('woot');
+        joins++;
 
-          if (joins == 2) {
-            setTimeout(function () {
-              connections.should.eql(3);
-              io.sockets.in('woot').json.send(123);
-            }, 20);
-          }
-        });
+        if (joins == 2) {
+          setTimeout(function () {
+            connections.should.eql(3);
+            io.sockets.in('woot').json.send(123);
+          }, 20);
+        }
       }
 
       socket.on('disconnect', function () {
@@ -868,16 +804,15 @@ module.exports = {
       connections++;
 
       if (connections != 3) {
-        socket.join('woot', function () {
-          joins++;
+        socket.join('woot');
+        joins++;
 
-          if (joins == 2) {
-            setTimeout(function () {
-              connections.should.eql(3);
-              io.sockets.in('woot').emit('locki');
-            }, 20);
-          }
-        });
+        if (joins == 2) {
+          setTimeout(function () {
+            connections.should.eql(3);
+            io.sockets.in('woot').emit('locki');
+          }, 20);
+        }
       }
 
       socket.on('disconnect', function () {
@@ -964,6 +899,59 @@ module.exports = {
     });
   },
 
+  'test leaving a room': function (done) {
+    var port = ++ports
+      , cl1 = client(port)
+      , cl2 = client(port)
+      , io = create(cl1)
+      , joins = 0
+      , disconnects = 0;
+
+    io.set('close timeout', 0);
+
+    io.sockets.on('connection', function (socket) {
+      socket.join('foo');
+      io.sockets.clients('foo').should.have.length(++joins);
+
+      socket.on('disconnect', function () {
+        socket.leave('foo');
+        socket.leave('foo');
+        socket.leave('foo');
+
+        io.sockets.clients('foo').should.have.length(--joins);
+
+        if (++disconnects == 2) {
+          io.server.close();
+          cl1.end();
+          cl2.end();
+          done();
+        }
+      })
+    });
+
+    cl1.handshake(function (sid) {
+      var ws1 = websocket(cl1, sid);
+      ws1.on('message', function (msg) {
+        if (!ws1.connected) {
+          msg.type.should.eql('connect');
+          ws1.connected = true;
+           ws1.finishClose();
+        }
+      });
+    });
+
+    cl2.handshake(function (sid) {
+      var ws2 = websocket(cl2, sid);
+      ws2.on('message', function (msg) {
+        if (!ws2.connected) {
+          msg.type.should.eql('connect');
+          ws2.connected = true;
+          ws2.finishClose();
+        }
+      });
+    });
+  },
+
   'test message with broadcast flag': function (done) {
     var port = ++ports
       , cl1 = client(port)
@@ -1268,8 +1256,9 @@ module.exports = {
     io.sockets.on('connection', function (socket) {
       connections++;
 
-      if (connections == 1)
+      if (connections == 1) {
         socket.join('losers');
+      }
 
       socket.on('trigger broadcast', function () {
         socket.broadcast.to('losers').send('boom');
@@ -1367,8 +1356,9 @@ module.exports = {
     io.sockets.on('connection', function (socket) {
       connections++;
 
-      if (connections == 1)
+      if (connections == 1) {
         socket.join('losers');
+      }
 
       socket.on('trigger broadcast', function () {
         socket.broadcast.json.to('losers').send({ hello: 'world' });
@@ -1548,6 +1538,234 @@ module.exports = {
         });
       });
     });
-  }
+  },
+
+  'test accessing handshake data from sockets': function (done) {
+    var cl = client(++ports)
+      , io = create(cl)
+      , ws;
+
+    io.sockets.on('connection', function (socket) {
+      (!!socket.handshake.address.address).should.be.true;
+      (!!socket.handshake.address.port).should.be.true;
+      socket.handshake.headers.host.should.equal('localhost');
+      socket.handshake.headers.connection.should.equal('keep-alive');
+      socket.handshake.time.should.match(/GMT/);
+
+      socket.on('disconnect', function () {
+        setTimeout(function () {
+          ws.finishClose();
+          cl.end();
+          io.server.close();
+          done();
+        }, 10);
+      });
+
+      socket.disconnect();
+    });
+
+    cl.handshake(function (sid) {
+      ws = websocket(cl, sid);
+      ws.on('message', function (msg) {
+        if (!ws.connected) {
+          msg.type.should.eql('connect');
+          ws.connected = true;
+        }
+      });
+    });
+  },
+
+  'test accessing the array of clients': function (done) {
+    var port = ++ports
+      , cl1 = client(port)
+      , cl2 = client(port)
+      , io = create(cl1)
+      , total = 2
+      , ws1, ws2;
+
+    io.sockets.on('connection', function (socket) {
+      socket.on('join ferrets', function () {
+        socket.join('ferrets');
+        socket.send('done');
+      });
+    });
+
+    function check() {
+      io.sockets.clients('ferrets').should.have.length(1);
+      io.sockets.clients('ferrets')[0].should.be.an.instanceof(sio.Socket);
+      io.sockets.clients('ferrets')[0].id.should.equal(ws1.sid);
+      io.sockets.clients().should.have.length(2);
+      io.sockets.clients()[0].should.be.an.instanceof(sio.Socket);
+      io.sockets.clients()[0].id.should.equal(ws1.sid);
+      io.sockets.clients()[1].should.be.an.instanceof(sio.Socket);
+      io.sockets.clients()[1].id.should.equal(ws2.sid);
+
+      ws1.finishClose();
+      ws2.finishClose();
+      cl1.end();
+      cl2.end();
+      io.server.close();
+      done();
+    };
+
+    cl1.handshake(function (sid) {
+      ws1 = websocket(cl1, sid);
+      ws1.sid = sid;
+      ws1.on('message', function (msg) {
+        if (!ws1.connected) {
+          msg.type.should.eql('connect');
+          ws1.connected = true;
+          ws1.packet({
+              type: 'event'
+            , name: 'join ferrets'
+            , endpoint: ''
+          });
+        } else {
+          cl2.handshake(function (sid) {
+            ws2 = websocket(cl2, sid);
+            ws2.sid = sid;
+            ws2.on('message', function (msg) {
+              if (!ws2.connected) {
+                msg.type.should.eql('connect');
+                ws2.connected = true;
+                check();
+              }
+            });
+          });
+        }
+      });
+    });
+  },
+
+  'test accessing handshake data from sockets on disconnect': function (done) {
+    var cl = client(++ports)
+      , io = create(cl)
+      , ws;
+
+    io.sockets.on('connection', function (socket) {
+      socket.on('disconnect', function () {
+
+      (!!socket.handshake.address.address).should.be.true;
+      (!!socket.handshake.address.port).should.be.true;
+      socket.handshake.headers.host.should.equal('localhost');
+      socket.handshake.headers.connection.should.equal('keep-alive');
+      socket.handshake.time.should.match(/GMT/);
+
+        setTimeout(function () {
+          ws.finishClose();
+          cl.end();
+          io.server.close();
+          done();
+        }, 10);
+      });
+
+      socket.disconnect();
+    });
+
+    cl.handshake(function (sid) {
+      ws = websocket(cl, sid);
+      ws.on('message', function (msg) {
+        if (!ws.connected) {
+          msg.type.should.eql('connect');
+          ws.connected = true;
+        }
+      });
+    });
+  },
+
+  'test for intentional and unintentional disconnects': function (done) {
+    var cl = client(++ports)
+      , io = create(cl)
+      , calls = 0
+      , ws;
+
+    function close () {
+      cl.end();
+      io.server.close();
+      ws.finishClose();
+      done();
+    }
+
+    io.configure(function () {
+      io.set('heartbeat interval', .05);
+      io.set('heartbeat timeout', .05);
+      io.set('close timeout', 0);
+    });
+
+    io.of('/foo').on('connection', function (socket) {
+      socket.on('disconnect', function (reason) {
+       reason.should.equal('packet');
+
+       if (++calls == 2) close();
+      });
+    });
+
+    io.of('/bar').on('connection', function (socket) {
+      socket.on('disconnect', function (reason) {
+        reason.should.equal('socket end');
+
+        if (++calls == 2) close();
+      });
+    });
+
+    cl.handshake(function (sid) {
+      var messages = 0;
+      ws = websocket(cl, sid);
+      ws.on('open', function () {
+        ws.packet({
+            type: 'connect'
+          , endpoint: '/foo'
+        });
+        ws.packet({
+            type: 'connect'
+          , endpoint: '/bar'
+        });
+      });
+
+      ws.on('message', function (packet) {
+        if (packet.type == 'connect') {
+          if (++messages === 3) {
+            ws.packet({ type: 'disconnect', endpoint:'/foo' });
+            ws.finishClose();
+          }
+        }
+      });
+    });
+  },
+
+  'test socket clean up': function (done) {
+    var cl = client(++ports)
+      , io = create(cl)
+      , ws;
+
+    io.sockets.on('connection', function (socket) {
+      var self = this
+        , id = socket.id;
+
+      socket.on('disconnect', function () {
+        setTimeout(function () {
+          var available = !!self.sockets[id];
+
+          available.should.be.false;
+          ws.finishClose();
+          cl.end();
+          io.server.close();
+          done();
+        }, 10);
+      });
+
+      socket.disconnect();
+    });
+
+    cl.handshake(function (sid) {
+      ws = websocket(cl, sid);
+      ws.on('message', function (msg) {
+        if (!ws.connected) {
+          msg.type.should.eql('connect');
+          ws.connected = true;
+        }
+      });
+    });
+  },
 
 };

test/transports.xhr-polling.test.js

@@ -113,7 +113,6 @@ module.exports = {
         --total || finish();
       });
 
-      // we rely on a small poll duration to close this request quickly
       cl.get('/socket.io/{protocol}/xhr-polling/' + sid, function (res, msgs) {
         res.statusCode.should.eql(200);
         msgs.should.have.length(1);
@@ -180,7 +179,6 @@ module.exports = {
         msgs.should.have.length(1);
         msgs[0].type.should.eql('connect');
 
-        // here we close the request instead of relying on a small poll timeout
         setTimeout(function () {
           cl.end();
         }, 10);
@@ -2637,4 +2635,82 @@ module.exports = {
     });
   },
 
+  'test emitting to closed clients': function (done) {
+    var cl = client(++ports)
+      , cl2 = client(ports)
+      , io = create(cl)
+      , connections = 0;
+
+    io.configure(function () {
+      io.set('close timeout', .1);
+    });
+
+    io.sockets.on('connection', function (socket) {
+      socket.send('a');
+    });
+
+    cl.handshake(function (sid) {
+      cl.get('/socket.io/{protocol}/xhr-polling/' + sid, function (res, packs) {
+        res.statusCode.should.equal(200);
+        packs.should.have.length(1);
+        packs[0].should.eql({ type: 'message', endpoint: '', data: 'a' });
+
+        cl2.handshake(function (sid2) {
+          cl2.get(
+              '/socket.io/{protocol}/xhr-polling/' + sid2
+            , function (res, packs) {
+                res.statusCode.should.equal(200);
+                packs.should.have.length(1);
+                packs[0].should.eql({ type: 'message', endpoint: '', data: 'a' });
+
+                io.sockets.emit('woot', 'b');
+
+                var total = 2;
+
+                cl.get(
+                    '/socket.io/{protocol}/xhr-polling/' + sid
+                  , function (res, packs) {
+                      res.statusCode.should.equal(200);
+                      packs.should.have.length(1);
+                      packs[0].should.eql({
+                          type: 'event'
+                        , endpoint: ''
+                        , name: 'woot'
+                        , args: ['b']
+                      });
+
+                      --total || finish();
+                    }
+                );
+
+                cl2.get(
+                    '/socket.io/{protocol}/xhr-polling/' + sid2
+                  , function (res, packs) {
+                      res.statusCode.should.equal(200);
+                      packs.should.have.length(1);
+                      packs[0].should.eql({
+                          type: 'event'
+                        , endpoint: ''
+                        , name: 'woot'
+                        , args: ['b']
+                      });
+
+                      --total || finish();
+                    }
+                );
+
+                function finish () {
+                  cl.end();
+                  cl2.end();
+                  io.server.close();
+                  done();
+                };
+              }
+          );
+        });
+        
+      });
+    });
+  }
+
 };