Implement masking as a transform stream.

StreamReader: Don't hang on to fully read buffers

.travis.yml

@@ -6,3 +6,6 @@ node_js:
   - "0.10"
   - "0.11"
 
+before_install:
+  - '[ "${TRAVIS_NODE_VERSION}" = "0.6" ] && npm conf set strict-ssl false || true'
+

CHANGELOG.md

@@ -0,0 +1,42 @@
+### 0.3.4 / 2014-05-08
+
+* Don't hold memory-leaking references to I/O buffers after they have been parsed
+
+### 0.3.3 / 2014-04-24
+
+* Correct the draft-76 status line reason phrase
+
+### 0.3.2 / 2013-12-29
+
+* Expand `maxLength` to cover sequences of continuation frames and `draft-{75,76}`
+* Decrease default maximum frame buffer size to 64MB
+* Stop parsing when the protocol enters a failure mode, to save CPU cycles
+
+### 0.3.1 / 2013-12-03
+
+* Add a `maxLength` option to limit allowed frame size
+* Don't pre-allocate a message buffer until the whole frame has arrived
+* Fix compatibility with Node v0.11 `HTTPParser`
+
+### 0.3.0 / 2013-09-09
+
+* Support client URLs with Basic Auth credentials
+
+### 0.2.2 / 2013-07-05
+
+* No functional changes, just updates to package.json
+
+### 0.2.1 / 2013-05-17
+
+* Export the isSecureRequest() method since faye-websocket relies on it
+* Queue sent messages in the client's initial state
+
+### 0.2.0 / 2013-05-12
+
+* Add API for setting and reading headers
+* Add Driver.server() method for getting a driver for TCP servers
+
+### 0.1.0 / 2013-05-04
+
+* First stable release
+

README.md

@@ -43,7 +43,7 @@ streams attached; one for incoming/outgoing messages and one for managing the
 wire protocol over an I/O stream. The full API is described below.
 
 
-### Server-side
+### Server-side with HTTP
 
 A Node webserver emits a special event for 'upgrade' requests, and this is
 where you should handle WebSockets. You first check whether the request is a
@@ -62,21 +62,66 @@ server.on('upgrade', function(request, socket, body) {
   var driver = websocket.http(request);
 
   driver.io.write(body);
-  socket.pipe(driver.io);
-  driver.io.pipe(socket);
+  socket.pipe(driver.io).pipe(socket);
 
   driver.messages.on('data', function(message) {
     console.log('Got a message', message);
   });
 
   driver.start();
-};
+});
 ```
 
 Note the line `driver.io.write(body)` - you must pass the `body` buffer to the
 socket driver in order to make certain versions of the protocol work.
 
 
+### Server-side with TCP
+
+You can also handle WebSocket connections in a bare TCP server, if you're not
+using an HTTP server and don't want to implement HTTP parsing yourself.
+
+The driver will emit a `connect` event when a request is received, and at this
+point you can detect whether it's a WebSocket and handle it as such. Here's an
+example using the Node `net` module:
+
+```js
+var net = require('net'),
+    websocket = require('websocket-driver');
+
+var server = net.createServer(function(connection) {
+  var driver = websocket.server();
+
+  driver.on('connect', function() {
+    if (websocket.isWebSocket(driver)) {
+      driver.start();
+    } else {
+      // handle other HTTP requests
+    }
+  });
+
+  driver.on('close', function() { connection.end() });
+  connection.on('error', function() {});
+
+  connection.pipe(driver.io).pipe(connection);
+
+  driver.messages.pipe(driver.messages);
+});
+
+server.listen(4180);
+```
+
+In the `connect` event, the driver gains several properties to describe the
+request, similar to a Node request object, such as `method`, `url` and
+`headers`. However you should remember it's not a real request object; you
+cannot write data to it, it only tells you what request data we parsed from the
+input.
+
+If the request has a body, it will be in the `driver.body` buffer, but only as
+much of the body as has been piped into the driver when the `connect` event
+fires.
+
+
 ### Client-side
 
 Similarly, to implement a WebSocket client you just need to make a driver by
@@ -91,8 +136,7 @@ var net = require('net'),
 var driver = websocket.client('ws://www.example.com/socket'),
     tcp = net.createConnection(80, 'www.example.com');
 
-tcp.pipe(driver.io);
-driver.io.pipe(tcp);
+tcp.pipe(driver.io).pipe(tcp);
 
 driver.messages.on('data', function(message) {
   console.log('Got a message', message);
@@ -103,6 +147,12 @@ tcp.on('connect', function() {
 });
 ```
 
+Client drivers have two additional properties for reading the HTTP data that
+was sent back by the server:
+
+* `driver.statusCode` - the integer value of the HTTP status code
+* `driver.headers` - an object containing the response headers
+
 
 ### Driver API
 
@@ -110,16 +160,21 @@ Drivers are created using one of the following methods:
 
 ```js
 driver = websocket.http(request, options)
+driver = websocket.server(options)
 driver = websocket.client(url, options)
 ```
 
 The `http` method returns a driver chosen using the headers from a Node HTTP
-request object. The `client` method always returns a driver for the RFC version
-of the protocol with masking enabled on outgoing frames.
+request object. The `server` method returns a driver that will parse an HTTP
+request and then decide which driver to use for it using the `http` method. The
+`client` method always returns a driver for the RFC version of the protocol
+with masking enabled on outgoing frames.
 
 The `options` argument is optional, and is an object. It may contain the
 following fields:
 
+* `maxLength` - the maximum allowed size of incoming message frames, in bytes.
+  The default value is `2^26 - 1`, or 1 byte short of 64 MiB.
 * `protocols` - an array of strings representing acceptable subprotocols for
   use over the socket. The driver will negotiate one of these to use via the
   `Sec-WebSocket-Protocol` header if supported by the other peer.
@@ -164,6 +219,12 @@ describing the error.
 Sets the callback to execute when the socket becomes closed. The `event` object
 has `code` and `reason` attributes.
 
+#### `driver.setHeader(name, value)`
+
+Sets a custom header to be sent as part of the handshake response, either from
+the server or from the client. Must be called before `start()`, since this is
+when the headers are serialized and sent.
+
 #### `driver.start()`
 
 Initiates the protocol by sending the handshake - either the response for a
@@ -224,7 +285,7 @@ after `emit('open')` has fired.
 
 (The MIT License)
 
-Copyright (c) 2010-2013 James Coglan
+Copyright (c) 2010-2014 James Coglan
 
 Permission is hereby granted, free of charge, to any person obtaining a copy of
 this software and associated documentation files (the 'Software'), to deal in

examples/tcp_server.js

@@ -0,0 +1,21 @@
+var net       = require('net'),
+    websocket = require('../lib/websocket/driver');
+
+var server = net.createServer(function(connection) {
+  var driver = websocket.server();
+
+  driver.on('connect', function() {
+    if (websocket.isWebSocket(driver)) driver.start();
+  });
+
+  driver.on('close', function() { connection.end() });
+  connection.on('error', function() {});
+
+  connection.pipe(driver.io);
+  driver.io.pipe(connection);
+
+  driver.messages.pipe(driver.messages);
+});
+
+server.listen(process.argv[2]);
+

lib/websocket/driver.js

@@ -4,45 +4,28 @@
 // * http://tools.ietf.org/html/draft-hixie-thewebsocketprotocol-76
 // * http://tools.ietf.org/html/draft-ietf-hybi-thewebsocketprotocol-17
 
-var Draft75 = require('./driver/draft75'),
-    Draft76 = require('./driver/draft76'),
-    Hybi    = require('./driver/hybi'),
-    Client  = require('./driver/client');
+var Client = require('./driver/client'),
+    Server = require('./driver/server');
 
 var Driver = {
-  isSecureRequest: function(request) {
-    if (request.headers['x-forwarded-proto']) {
-      return request.headers['x-forwarded-proto'] === 'https';
-    } else {
-      return (request.connection && request.connection.authorized !== undefined) ||
-             (request.socket && request.socket.secure);
-    }
-  },
-
-  determineUrl: function(request) {
-    var scheme = this.isSecureRequest(request) ? 'wss:' : 'ws:';
-    return scheme + '//' + request.headers.host + request.url;
-  },
-
   client: function(url, options) {
     options = options || {};
     if (options.masking === undefined) options.masking = true;
     return new Client(url, options);
   },
 
-  http: function(request, options) {
+  server: function(options) {
     options = options || {};
     if (options.requireMasking === undefined) options.requireMasking = true;
+    return new Server(options);
+  },
 
-    var headers = request.headers,
-        url     = this.determineUrl(request);
+  http: function() {
+    return Server.http.apply(Server, arguments);
+  },
 
-    if (headers['sec-websocket-version'])
-      return new Hybi(request, url, options);
-    else if (headers['sec-websocket-key1'])
-      return new Draft76(request, url, options);
-    else
-      return new Draft75(request, url, options);
+  isSecureRequest: function(request) {
+    return Server.isSecureRequest(request);
   },
 
   isWebSocket: function(request) {

lib/websocket/driver/base.js

@@ -1,50 +1,66 @@
 var Emitter = require('events').EventEmitter,
     util    = require('util'),
-    streams = require('../streams');
+    streams = require('../streams'),
+    Headers = require('./headers');
 
 var Base = function(request, url, options) {
   Emitter.call(this);
 
   this._request   = request;
   this._options   = options || {};
+  this._maxLength = this._options.maxLength || this.MAX_LENGTH;
+  this.__headers  = new Headers();
   this.__queue    = [];
   this.readyState = 0;
   this.url        = url;
 
-  var self = this;
-
   this.io = new streams.IO(this);
   this.messages = new streams.Messages(this);
-
-  // Protocol errors are informational and do not have to be handled
-  this.messages.on('error', function() {});
-
-  this.on('message', function(event) {
-    var messages = self.messages;
-    if (messages.readable) messages.emit('data', event.data);
-  });
-
-  this.on('error', function(error) {
-    var messages = self.messages;
-    if (messages.readable) messages.emit('error', error);
-  });
-
-  this.on('close', function() {
-    var messages = self.messages;
-    if (!messages.readable) return;
-    messages.readable = messages.writable = false;
-    messages.emit('end');
-  });
+  this._bindEventListeners();
 };
 util.inherits(Base, Emitter);
 
 var instance = {
+  // This is 64MB, small enough for an average VPS to handle without
+  // crashing from process out of memory
+  MAX_LENGTH: 0x3ffffff,
+
   STATES: ['connecting', 'open', 'closing', 'closed'],
 
+  _bindEventListeners: function() {
+    var self = this;
+
+    // Protocol errors are informational and do not have to be handled
+    this.messages.on('error', function() {});
+
+    this.on('message', function(event) {
+      var messages = self.messages;
+      if (messages.readable) messages.emit('data', event.data);
+    });
+
+    this.on('error', function(error) {
+      var messages = self.messages;
+      if (messages.readable) messages.emit('error', error);
+    });
+
+    this.on('close', function() {
+      var messages = self.messages;
+      if (!messages.readable) return;
+      messages.readable = messages.writable = false;
+      messages.emit('end');
+    });
+  },
+
   getState: function() {
     return this.STATES[this.readyState] || null;
   },
 
+  setHeader: function(name, value) {
+    if (this.readyState > 0) return false;
+    this.__headers.set(name, value);
+    return true;
+  },
+
   start: function() {
     if (this.readyState !== 0) return false;
     this._write(this._handshakeResponse());
@@ -93,6 +109,8 @@ for (var key in instance)
   Base.prototype[key] = instance[key];
 
 
+Base.ConnectEvent = function() {};
+
 Base.OpenEvent = function() {};
 
 Base.CloseEvent = function(code, reason) {

lib/websocket/driver/client.js

@@ -1,6 +1,6 @@
-var HTTPParser = process.binding('http_parser').HTTPParser,
-    url        = require('url'),
+var url        = require('url'),
     util       = require('util'),
+    HttpParser = require('./http_parser'),
     Base       = require('./base'),
     Hybi       = require('./hybi');
 
@@ -11,37 +11,13 @@ var Client = function(url, options) {
   this.readyState = -1;
   this._key       = Client.generateKey();
   this._accept    = Hybi.generateAccept(this._key);
-
-  this._http     = new HTTPParser(HTTPParser.RESPONSE || 'response');
-  this._node     = HTTPParser.RESPONSE ? 6 : 4;
-  this._complete = false;
-  this._headers  = {};
-
-  var currentHeader = null,
-      self          = this;
-
-  this._http.onHeaderField = function(b, start, length) {
-    currentHeader = b.toString('utf8', start, start + length);
-  };
-  this._http.onHeaderValue = function(b, start, length) {
-    self._headers[currentHeader] = b.toString('utf8', start, start + length);
-  };
-  this._http.onHeadersComplete = function(info) {
-    self._status = info.statusCode;
-    var headers = info.headers;
-    if (!headers) return;
-    for (var i = 0, n = headers.length; i < n; i += 2)
-      self._headers[headers[i]] = headers[i+1];
-  };
-  this._http.onMessageComplete = function() {
-    self._complete = true;
-  };
+  this._http      = new HttpParser('response');
 };
 util.inherits(Client, Hybi);
 
 Client.generateKey = function() {
   var buffer = new Buffer(16), i = buffer.length;
-  while (i--) buffer[i] = Math.floor(Math.random() * 256);
+  while (i--) buffer[i] = ~~(Math.random() * 256);
   return buffer.toString('base64');
 };
 
@@ -56,11 +32,11 @@ var instance = {
   parse: function(data) {
     if (this.readyState > 0) return Hybi.prototype.parse.call(this, data);
 
-    var consumed = this._http.execute(data, 0, data.length),
-        offset   = (this._node < 6) ? 1 : 0; 
-
-    if (consumed <= data.length) this._validateHandshake();
-    if (consumed < data.length)  this.parse(data.slice(consumed + offset));
+    this._http.parse(data);
+    if (!this._http.isComplete()) return;
+    
+    this._validateHandshake();
+    this.parse(this._http.body);
   },
 
   _handshakeRequest: function() {
@@ -77,7 +53,10 @@ var instance = {
     if (this._protocols.length > 0)
       headers.push('Sec-WebSocket-Protocol: ' + this._protocols.join(', '));
 
-    return new Buffer(headers.concat('','').join('\r\n'), 'utf8');
+    if (uri.auth)
+      headers.push('Authorization: Basic ' + new Buffer(uri.auth, 'utf8').toString('base64'));
+
+    return new Buffer(headers.concat(this.__headers.toString(), '').join('\r\n'), 'utf8');
   },
 
   _failHandshake: function(message) {
@@ -88,13 +67,17 @@ var instance = {
   },
 
   _validateHandshake: function() {
-    if (this._status !== 101)
-      return this._failHandshake('Unexpected response code: ' + this._status);
+    this.statusCode = this._http.statusCode;
+    this.headers    = this._http.headers;
 
-    var upgrade    = this._headers.Upgrade || '',
-        connection = this._headers.Connection || '',
-        accept     = this._headers['Sec-WebSocket-Accept'] || '',
-        protocol   = this._headers['Sec-WebSocket-Protocol'] || '';
+    if (this._http.statusCode !== 101)
+      return this._failHandshake('Unexpected response code: ' + this._http.statusCode);
+
+    var headers    = this._http.headers,
+        upgrade    = headers['upgrade'] || '',
+        connection = headers['connection'] || '',
+        accept     = headers['sec-websocket-accept'] || '',
+        protocol   = headers['sec-websocket-protocol'] || '';
 
     if (upgrade === '')
       return this._failHandshake("'Upgrade' header is missing");

lib/websocket/driver/draft75.js

@@ -9,7 +9,16 @@ var Draft75 = function(request, url, options) {
 util.inherits(Draft75, Base);
 
 var instance = {
+  close: function() {
+    if (this.readyState === 3) return false;
+    this.readyState = 3;
+    this.emit('close', new Base.CloseEvent(null, null));
+    return true;
+  },
+
   parse: function(buffer) {
+    if (this.readyState > 1) return;
+
     var data, message, value;
     for (var i = 0, n = buffer.length; i < n; i++) {
       data = buffer[i];
@@ -29,8 +38,7 @@ var instance = {
           this._length = value + 128 * this._length;
 
           if (this._closing && this._length === 0) {
-            this.readyState = 3;
-            this.emit('close', new Base.CloseEvent(null, null));
+            return this.close();
           }
           else if ((0x80 & data) !== 0x80) {
             if (this._length === 0) {
@@ -56,6 +64,7 @@ var instance = {
                 this._stage = 0;
             } else {
               this._buffer.push(data);
+              if (this._buffer.length > this._maxLength) return this.close();
             }
           }
           break;
@@ -84,6 +93,7 @@ var instance = {
                       'Connection: Upgrade\r\n' +
                       'WebSocket-Origin: ' + this._request.headers.origin + '\r\n' +
                       'WebSocket-Location: ' + this.url + '\r\n' +
+                      this.__headers.toString() +
                       '\r\n',
                       'utf8');
   },

lib/websocket/driver/draft76.js

@@ -48,11 +48,12 @@ var instance = {
   },
 
   _handshakeResponse: function() {
-    return new Buffer('HTTP/1.1 101 Web Socket Protocol Handshake\r\n' +
+    return new Buffer('HTTP/1.1 101 WebSocket Protocol Handshake\r\n' +
                       'Upgrade: WebSocket\r\n' +
                       'Connection: Upgrade\r\n' +
                       'Sec-WebSocket-Origin: ' + this._request.headers.origin + '\r\n' +
                       'Sec-WebSocket-Location: ' + this.url + '\r\n' +
+                      this.__headers.toString() +
                       '\r\n',
                       'binary');
   },

lib/websocket/driver/headers.js

@@ -0,0 +1,30 @@
+var Headers = function() {
+  this._sent  = {};
+  this._lines = [];
+};
+
+Headers.prototype.ALLOWED_DUPLICATES = ['set-cookie', 'set-cookie2', 'warning', 'www-authenticate']
+
+Headers.prototype.set = function(name, value) {
+  if (value === undefined) return;
+
+  name = this._strip(name);
+  value = this._strip(value);
+
+  var key = name.toLowerCase();
+  if (!this._sent.hasOwnProperty(key) || this.ALLOWED_DUPLICATES.indexOf(key) < 0) {
+    this._sent[key] = true;
+    this._lines.push(name + ': ' + value + '\r\n');
+  }
+};
+
+Headers.prototype.toString = function() {
+  return this._lines.join('');
+};
+
+Headers.prototype._strip = function(string) {
+  return string.toString().replace(/^ */, '').replace(/ *$/, '');
+};
+
+module.exports = Headers;
+

lib/websocket/driver/http_parser.js

@@ -0,0 +1,81 @@
+var HTTPParser = process.binding('http_parser').HTTPParser,
+    version    = HTTPParser.RESPONSE ? 6 : 4;
+
+var HttpParser = function(type) {
+  if (type === 'request')
+    this._parser = new HTTPParser(HTTPParser.REQUEST || 'request');
+  else
+    this._parser = new HTTPParser(HTTPParser.RESPONSE || 'response');
+
+  this._type     = type;
+  this._complete = false;
+  this.headers   = {};
+
+  var current = null,
+      self    = this;
+
+  this._parser.onHeaderField = function(b, start, length) {
+    current = b.toString('utf8', start, start + length).toLowerCase();
+  };
+
+  this._parser.onHeaderValue = function(b, start, length) {
+    self.headers[current] = b.toString('utf8', start, start + length);
+  };
+
+  this._parser.onHeadersComplete = this._parser[HTTPParser.kOnHeadersComplete] = function(info) {
+    self.method     = (typeof info.method === 'number') ? HttpParser.METHODS[info.method] : info.method;
+    self.statusCode = info.statusCode;
+    self.url        = info.url;
+    
+    var headers = info.headers;
+    if (!headers) return;
+
+    for (var i = 0, n = headers.length; i < n; i += 2)
+      self.headers[headers[i].toLowerCase()] = headers[i+1];
+  };
+
+  this._parser.onMessageComplete = this._parser[HTTPParser.kOnMessageComplete] = function() {
+    self._complete = true;
+  };
+};
+
+HttpParser.METHODS = {
+  0:  'DELETE',
+  1:  'GET',
+  2:  'HEAD',
+  3:  'POST',
+  4:  'PUT',
+  5:  'CONNECT',
+  6:  'OPTIONS',
+  7:  'TRACE',
+  8:  'COPY',
+  9:  'LOCK',
+  10: 'MKCOL',
+  11: 'MOVE',
+  12: 'PROPFIND',
+  13: 'PROPPATCH',
+  14: 'SEARCH',
+  15: 'UNLOCK',
+  16: 'REPORT',
+  17: 'MKACTIVITY',
+  18: 'CHECKOUT',
+  19: 'MERGE',
+  24: 'PATCH'
+};
+
+HttpParser.prototype.isComplete = function() {
+  return this._complete;
+};
+
+HttpParser.prototype.parse = function(data) {
+  var offset   = (version < 6) ? 1 : 0,
+      consumed = this._parser.execute(data, 0, data.length) + offset;
+
+  if (this._complete)
+    this.body = (consumed < data.length)
+              ? data.slice(consumed)
+              : new Buffer(0);
+};
+
+module.exports = HttpParser;
+

lib/websocket/driver/hybi.js

@@ -1,14 +1,15 @@
-var crypto   = require('crypto'),
-    util     = require('util'),
-    Base     = require('./base'),
-    Reader   = require('./hybi/stream_reader');
+var crypto = require('crypto'),
+    util   = require('util'),
+    Base   = require('./base'),
+    Concat = require('./hybi/concat'),
+    Mask   = require('./hybi/mask'),
+    Reader = require('./hybi/stream_reader');
 
 var Hybi = function(request, url, options) {
   Base.apply(this, arguments);
   this._reset();
 
-  this._reader    = new Reader();
-  this._stage     = 0;
+  this._reader    = new Reader({context: this});
   this._masking   = this._options.masking;
   this._protocols = this._options.protocols || [];
 
@@ -22,19 +23,11 @@ var Hybi = function(request, url, options) {
     var version = this._request.headers['sec-websocket-version'];
     this.version = 'hybi-' + version;
   }
+
+  this._reader.read(1, this._parseOpcode);
 };
 util.inherits(Hybi, Base);
 
-Hybi.mask = function(payload, mask, offset) {
-  if (mask.length === 0) return payload;
-  offset = offset || 0;
-
-  for (var i = 0, n = payload.length - offset; i < n; i++) {
-    payload[offset + i] = payload[offset + i] ^ mask[i % 4];
-  }
-  return payload;
-};
-
 Hybi.generateAccept = function(key) {
   var sha1 = crypto.createHash('sha1');
   sha1.update(key + Hybi.GUID);
@@ -66,7 +59,6 @@ var instance = {
   FRAGMENTED_OPCODES: [0, 1, 2],
   OPENING_OPCODES:    [1, 2],
 
-  MAX_LENGTH: Math.pow(2, 53) - 1,
   TWO_POWERS: [0, 1, 2, 3, 4, 5, 6, 7].map(function(n) { return Math.pow(2, 8 * n) }),
 
   ERRORS: {
@@ -89,47 +81,11 @@ var instance = {
   UTF8_MATCH: /^([\x00-\x7F]|[\xC2-\xDF][\x80-\xBF]|\xE0[\xA0-\xBF][\x80-\xBF]|[\xE1-\xEC\xEE\xEF][\x80-\xBF]{2}|\xED[\x80-\x9F][\x80-\xBF]|\xF0[\x90-\xBF][\x80-\xBF]{2}|[\xF1-\xF3][\x80-\xBF]{3}|\xF4[\x80-\x8F][\x80-\xBF]{2})*$/,
 
   parse: function(data) {
-    this._reader.put(data);
-    var buffer = true;
-    while (buffer) {
-      switch (this._stage) {
-        case 0:
-          buffer = this._reader.read(1);
-          if (buffer) this._parseOpcode(buffer[0]);
-          break;
-
-        case 1:
-          buffer = this._reader.read(1);
-          if (buffer) this._parseLength(buffer[0]);
-          break;
-
-        case 2:
-          buffer = this._reader.read(this._lengthSize);
-          if (buffer) this._parseExtendedLength(buffer);
-          break;
-
-        case 3:
-          buffer = this._reader.read(4);
-          if (buffer) {
-            this._mask  = buffer;
-            this._stage = 4;
-          }
-          break;
-
-        case 4:
-          buffer = this._reader.read(this._length);
-          if (buffer) {
-            this._payload = buffer;
-            this._emitFrame();
-            this._stage = 0;
-          }
-          break;
-      }
-    }
+    this._reader.write(data);
   },
 
   frame: function(data, type, code) {
-    if (this.readyState === 0) return this._queue([data, type, code]);
+    if (this.readyState <= 0) return this._queue([data, type, code]);
     if (this.readyState !== 1) return false;
 
     if (data instanceof Array) data = new Buffer(data);
@@ -152,31 +108,31 @@ var instance = {
       frame[1] = masked | length;
     } else if (length <= 65535) {
       frame[1] = masked | 126;
-      frame[2] = Math.floor(length / 256);
+      frame[2] = ~~(length / 256);
       frame[3] = length & BYTE;
     } else {
       frame[1] = masked | 127;
-      frame[2] = Math.floor(length / Math.pow(2,56)) & BYTE;
-      frame[3] = Math.floor(length / Math.pow(2,48)) & BYTE;
-      frame[4] = Math.floor(length / Math.pow(2,40)) & BYTE;
-      frame[5] = Math.floor(length / Math.pow(2,32)) & BYTE;
-      frame[6] = Math.floor(length / Math.pow(2,24)) & BYTE;
-      frame[7] = Math.floor(length / Math.pow(2,16)) & BYTE;
-      frame[8] = Math.floor(length / Math.pow(2,8))  & BYTE;
+      frame[2] = ~~(length / Math.pow(2, 56)) & BYTE;
+      frame[3] = ~~(length / Math.pow(2, 48)) & BYTE;
+      frame[4] = ~~(length / Math.pow(2, 40)) & BYTE;
+      frame[5] = ~~(length / Math.pow(2, 32)) & BYTE;
+      frame[6] = ~~(length / Math.pow(2, 24)) & BYTE;
+      frame[7] = ~~(length / Math.pow(2, 16)) & BYTE;
+      frame[8] = ~~(length / Math.pow(2, 8))  & BYTE;
       frame[9] = length & BYTE;
     }
 
     if (code) {
-      frame[offset]   = Math.floor(code / 256) & BYTE;
-      frame[offset+1] = code & BYTE;
+      frame[offset]     = ~~(code / 256) & BYTE;
+      frame[offset + 1] = code & BYTE;
     }
     buffer.copy(frame, offset + insert);
 
     if (this._masking) {
-      mask = [Math.floor(Math.random() * 256), Math.floor(Math.random() * 256),
-              Math.floor(Math.random() * 256), Math.floor(Math.random() * 256)];
+      mask = [~~(Math.random() * 256), ~~(Math.random() * 256),
+              ~~(Math.random() * 256), ~~(Math.random() * 256)];
       new Buffer(mask).copy(frame, header);
-      Hybi.mask(frame, mask, offset);
+      Mask.mask(frame, mask, offset);
     }
 
     this._write(frame);
@@ -239,12 +195,13 @@ var instance = {
       }
     }
 
-    return new Buffer(headers.concat('','').join('\r\n'), 'utf8');
+    return new Buffer(headers.concat(this.__headers.toString(), '').join('\r\n'), 'utf8');
   },
 
   _shutdown: function(code, reason) {
     this.frame(reason, 'close', code);
     this.readyState = 3;
+    this._reader.end();
     this.emit('close', new Base.CloseEvent(code, reason));
   },
 
@@ -253,7 +210,9 @@ var instance = {
     this._shutdown(this.ERRORS[type], message);
   },
 
-  _parseOpcode: function(data) {
+  _parseOpcode: function(buffer) {
+    var data = buffer[0];
+
     var rsvs = [this.RSV1, this.RSV2, this.RSV3].map(function(rsv) {
       return (data & rsv) === rsv;
     });
@@ -264,10 +223,9 @@ var instance = {
           ', reserved2 = ' + (rsvs[1] ? 1 : 0) +
           ', reserved3 = ' + (rsvs[2] ? 1 : 0));
 
-    this._final   = (data & this.FIN) === this.FIN;
-    this._opcode  = (data & this.OPCODE);
-    this._mask    = [];
-    this._payload = [];
+    this._final  = (data & this.FIN) === this.FIN;
+    this._opcode = (data & this.OPCODE);
+    this._mask   = null;
 
     if (this.OPCODE_CODES.indexOf(this._opcode) < 0)
       return this._fail('protocol_error', 'Unrecognized frame opcode: ' + this._opcode);
@@ -278,10 +236,12 @@ var instance = {
     if (this._mode && this.OPENING_OPCODES.indexOf(this._opcode) >= 0)
       return this._fail('protocol_error', 'Received new data frame but previous continuous frame is unfinished');
 
-    this._stage = 1;
+    this._reader.read(1, this._parseLength);
   },
 
-  _parseLength: function(data) {
+  _parseLength: function(buffer) {
+    var data = buffer[0];
+
     this._masked = (data & this.MASK) === this.MASK;
     if (this._requireMasking && !this._masked)
       return this._fail('unacceptable', 'Received unmasked frame but masking is required');
@@ -289,11 +249,11 @@ var instance = {
     this._length = (data & this.LENGTH);
 
     if (this._length >= 0 && this._length <= 125) {
-      this._stage = this._masked ? 3 : 4;
+      if (!this._checkFrameLength()) return;
+      this._readMask();
     } else {
-      this._lengthBuffer = [];
-      this._lengthSize   = (this._length === 126 ? 2 : 8);
-      this._stage        = 2;
+      var lengthSize = (this._length === 126 ? 2 : 8);
+      this._reader.read(lengthSize, this._parseExtendedLength);
     }
   },
 
@@ -303,21 +263,47 @@ var instance = {
     if (this.FRAGMENTED_OPCODES.indexOf(this._opcode) < 0 && this._length > 125)
       return this._fail('protocol_error', 'Received control frame having too long payload: ' + this._length);
 
-    if (this._length > this.MAX_LENGTH)
-      return this._fail('too_large', 'WebSocket frame length too large');
+    if (!this._checkFrameLength()) return;
 
-    this._stage  = this._masked ? 3 : 4;
+    this._readMask();
   },
 
-  _emitFrame: function() {
-    var payload = Hybi.mask(this._payload, this._mask),
-        opcode  = this._opcode;
+  _checkFrameLength: function() {
+    if (this.__blength + this._length > this._maxLength) {
+      this._fail('too_large', 'WebSocket frame length too large');
+      return false;
+    } else {
+      return true;
+    }
+  },
+
+  _readMask: function() {
+    if (this._masked)
+      this._reader.read(4, function(buffer) {
+        this._mask = new Mask(buffer);
+        this._readPayload();
+      });
+    else
+      this._readPayload();
+  },
+
+  _readPayload: function() {
+    var stream = this._reader.fork(this._length);
+    if (this._mask) {
+      stream.pipe(this._mask);
+      stream = this._mask;
+    }
+    stream.pipe(new Concat(this._emitFrame, this));
+  },
+
+  _emitFrame: function(payload) {
+    var opcode = this._opcode;
 
     if (opcode === this.OPCODES.continuation) {
       if (!this._mode) return this._fail('protocol_error', 'Received unexpected continuation frame');
       this._buffer(payload);
       if (this._final) {
-        var message = new Buffer(this.__buffer);
+        var message = Concat.concatBuffers(this.__buffer, this.__blength);
         if (this._mode === 'text') message = this._encode(message);
         this._reset();
         if (message === null)
@@ -371,16 +357,19 @@ var instance = {
       delete callbacks[message];
       if (callback) callback()
     }
+
+    this._reader.read(1, this._parseOpcode);
   },
 
   _buffer: function(fragment) {
-    for (var i = 0, n = fragment.length; i < n; i++)
-      this.__buffer.push(fragment[i]);
+    this.__buffer.push(fragment);
+    this.__blength += fragment.length;
   },
 
   _reset: function() {
-    this._mode    = null;
-    this.__buffer = [];
+    this._mode     = null;
+    this.__buffer  = [];
+    this.__blength = 0;
   },
 
   _encode: function(buffer) {

lib/websocket/driver/hybi/concat.js

@@ -0,0 +1,45 @@
+var Stream = require('stream').Stream,
+    util   = require('util');
+
+var Concat = function(callback, context) {
+  this._callback = callback;
+  this._context  = context;
+  this._chunks   = [];
+  this._size     = 0;
+  this.writable  = true;
+};
+util.inherits(Concat, Stream);
+
+Concat.prototype.write = function(buffer) {
+  if (!this.writable) return false;
+  this._chunks.push(buffer);
+  this._size += buffer.length;
+  return true;
+};
+
+Concat.prototype.end = function(buffer) {
+  if (buffer) this.write(buffer);
+  this.writable = false;
+  this._callback.call(this._context, Concat.concatBuffers(this._chunks, this._size));
+};
+
+Concat.concatBuffers = function(chunks, size) {
+  if (size === undefined) {
+    size = 0;
+    var c = chunks.length;
+    while (c--) size += chunks[c].length;
+  }
+
+  var concat = new Buffer(size),
+      offset = 0;
+
+  for (var i = 0, n = chunks.length; i < n; i++) {
+    chunks[i].copy(concat, offset);
+    offset += chunks[i].length;
+  }
+  return concat;
+};
+
+Concat.prototype.destroy = function() {};
+
+module.exports = Concat;

lib/websocket/driver/hybi/mask.js

@@ -0,0 +1,43 @@
+var Stream = require('stream').Stream,
+    util   = require('util');
+
+var Mask = function(bytes) {
+  this.readable = this.writable = true;
+
+  this._bytes = bytes;
+  this._index = 0;
+};
+util.inherits(Mask, Stream);
+
+Mask.mask = function(payload, mask, offset, index) {
+  offset = offset || 0;
+  index  = index  || 0;
+
+  for (var i = 0, n = payload.length - offset; i < n; i++)
+    payload[offset + i] ^= mask[(index + i) % 4];
+};
+
+Mask.prototype.write = function(chunk) {
+  Mask.mask(chunk, this._bytes, 0, this._index);
+  this._index = (this._index + chunk.length) % 4;
+  this.emit('data', chunk);
+  return !this._paused;
+};
+
+Mask.prototype.end = function(chunk) {
+  if (chunk) this.write(chunk);
+  this.readable = this.writable = false;
+  this.emit('end');
+};
+
+Mask.prototype.pause = function() {
+  this._paused = true;
+};
+
+Mask.prototype.resume = function() {
+  this._paused = false;
+  this.emit('drain');
+};
+
+module.exports = Mask;
+

lib/websocket/driver/hybi/stream_reader.js

@@ -1,41 +1,136 @@
-var StreamReader = function() {
-  this._queue  = [];
-  this._cursor = 0;
-};
+var Stream = require('stream').Stream,
+    Concat = require('./concat'),
+    util   = require('util');
 
-StreamReader.prototype.read = function(bytes) {
-  return this._readBuffer(bytes);
-};
+var defer = (typeof setImmediate === 'function')
+          ? setImmediate
+          : process.nextTick;
+
+var StreamReader = function(options, parent) {
+  this.readable   = !!parent;
+  this.writable   = !parent;
+  this._streams   = [];
+  this._context   = options.context;
+  this._parent    = parent;
+  this._queue     = [];
+  this._queueSize = 0;
+  this._cursor    = 0;
+};
+util.inherits(StreamReader, Stream);
+
+StreamReader.prototype.write = function(buffer) {
+  if (!this.writable) return false;
+  if (!buffer || buffer.length === 0) return !this._paused;
 
-StreamReader.prototype.put = function(buffer) {
-  if (!buffer || buffer.length === 0) return;
   if (!buffer.copy) buffer = new Buffer(buffer);
+
   this._queue.push(buffer);
+  this._queueSize += buffer.length;
+  this._flush();
+
+  return !this._paused;
 };
 
-StreamReader.prototype._readBuffer = function(length) {
-  var buffer = new Buffer(length),
-      queue  = this._queue,
+StreamReader.prototype.end = function(buffer) {
+  if (buffer) this.write(buffer);
+  this.writable = false;
+
+  for (var i = 0, n = this._streams.length; i < n; i++) {
+    this._streams[i].emit('end');
+    this._streams[i].readable = false;
+  }
+
+  this._context = this._streams = this._queue = [];
+};
+
+StreamReader.prototype.pause = function() {
+  this._paused = true;
+  if (this._parent) this._parent.pause();
+};
+
+StreamReader.prototype.resume = function() {
+  this._paused = false;
+  this.emit('drain');
+  if (this._parent) this._parent.resume();
+};
+
+StreamReader.prototype.fork = function(length) {
+  if (!this.writable) return null;
+
+  var stream = new StreamReader({context: this._context}, this),
+      self   = this;
+
+  stream._remaining = length;
+  this._streams.push(stream);
+  defer(function() { self._flush() });
+
+  return stream;
+};
+
+StreamReader.prototype.read = function(length, callback) {
+  if (!this.writable) return;
+
+  if (this._queueSize >= length)
+    return callback.call(this._context, this._readBytes(length));
+
+  this.fork(length).pipe(new Concat(callback, this._context));
+};
+
+StreamReader.prototype._flush = function() {
+  var streams = this._streams, stream, size, buffer;
+
+  while (streams.length > 0) {
+    stream = streams[0];
+    size   = Math.min(stream._remaining, this._queueSize);
+    buffer = this._readBytes(size);
+
+    if (size > 0) stream.emit('data', buffer);
+    stream._remaining -= size;
+
+    if (stream._remaining > 0) break;
+
+    stream.readable = false;
+    stream.emit('end');
+    streams.shift();
+  }
+};
+
+StreamReader.prototype._readBytes = function(length) {
+  var queue  = this._queue,
       remain = length,
       n      = queue.length,
+      first  = queue[0],
       i      = 0,
-      chunk, offset, size;
+      buffer, chunk, size;
 
-  if (remain === 0) return buffer;
+  if (length === 0) return new Buffer(0);
+
+  if (remain <= first.length - this._cursor) {
+    buffer = first.slice(this._cursor, this._cursor + remain);
+
+    this._queueSize -= remain;
+    this._cursor     = (this._cursor + remain) % first.length;
+    if (this._cursor === 0) this._queue.shift();
+
+    return buffer;
+  }
+
+  buffer = new Buffer(length);
 
   while (remain > 0 && i < n) {
     chunk = queue[i];
-    offset = (i === 0) ? this._cursor : 0;
-    size = Math.min(remain, chunk.length - offset);
-    chunk.copy(buffer, length - remain, offset, offset + size);
-    remain -= size;
+    size  = Math.min(remain, chunk.length - this._cursor);
+
+    chunk.copy(buffer, length - remain, this._cursor, this._cursor + size);
+
+    remain          -= size;
+    this._queueSize -= size;
+    this._cursor     = (this._cursor + size) % chunk.length;
+
     i += 1;
   }
 
-  if (remain > 0) return null;
-
-  queue.splice(0, i-1);
-  this._cursor = (i === 1 ? this._cursor : 0) + size;
+  queue.splice(0, this._cursor === 0 ? i : i - 1);
 
   return buffer;
 };

lib/websocket/driver/server.js

@@ -0,0 +1,104 @@
+var util       = require('util'),
+    HttpParser = require('./http_parser'),
+    Base       = require('./base'),
+    Draft75    = require('./draft75'),
+    Draft76    = require('./draft76'),
+    Hybi       = require('./hybi');
+
+var Server = function(options) {
+  Base.call(this, null, null, options);
+  this._http = new HttpParser('request');
+};
+util.inherits(Server, Base);
+
+var instance = {
+  EVENTS: ['open', 'message', 'error', 'close'],
+
+  _bindEventListeners: function() {
+    this.messages.on('error', function() {});
+    this.on('error', function() {});
+  },
+
+  parse: function(data) {
+    if (this._delegate) return this._delegate.parse(data);
+
+    this._http.parse(data);
+    if (!this._http.isComplete()) return;
+
+    this.method  = this._http.method;
+    this.url     = this._http.url;
+    this.headers = this._http.headers;
+    this.body    = this._http.body;
+
+    var self = this;
+    this._delegate = Server.http(this, this._options);
+    this._delegate.messages = this.messages;
+    this._delegate.io = this.io;
+
+    this._delegate.on('open', function() { self._open() });
+    this.EVENTS.forEach(function(event) {
+      this._delegate.on(event, function(e) { self.emit(event, e) });
+    }, this);
+
+    this.parse(this._http.body);
+    this.emit('connect', new Base.ConnectEvent());
+  },
+
+  _open: function() {
+    this.__queue.forEach(function(msg) {
+      this._delegate[msg[0]].apply(this._delegate, msg[1]);
+    }, this);
+    this.__queue = [];
+  }
+};
+
+['setHeader', 'start', 'state', 'frame', 'text', 'binary', 'ping', 'close'].forEach(function(method) {
+  instance[method] = function() {
+    if (this._delegate) {
+      return this._delegate[method].apply(this._delegate, arguments);
+    } else {
+      this.__queue.push([method, arguments]);
+      return true;
+    }
+  };
+});
+
+for (var key in instance)
+  Server.prototype[key] = instance[key];
+
+Server.isSecureRequest = function(request) {
+  if (request.connection && request.connection.authorized !== undefined) return true;
+  if (request.socket && request.socket.secure) return true;
+
+  var headers = request.headers;
+  if (!headers) return false;
+  if (headers['https'] === 'on') return true;
+  if (headers['x-forwarded-ssl'] === 'on') return true;
+  if (headers['x-forwarded-scheme'] === 'https') return true;
+  if (headers['x-forwarded-proto'] === 'https') return true;
+
+  return false;
+};
+
+Server.determineUrl = function(request) {
+  var scheme = this.isSecureRequest(request) ? 'wss:' : 'ws:';
+  return scheme + '//' + request.headers.host + request.url;
+};
+
+Server.http = function(request, options) {
+  options = options || {};
+  if (options.requireMasking === undefined) options.requireMasking = true;
+
+  var headers = request.headers,
+      url     = this.determineUrl(request);
+
+  if (headers['sec-websocket-version'])
+    return new Hybi(request, url, options);
+  else if (headers['sec-websocket-key1'])
+    return new Draft76(request, url, options);
+  else
+    return new Draft75(request, url, options);
+};
+
+module.exports = Server;
+

package.json

@@ -3,24 +3,19 @@
 , "homepage"        : "http://github.com/faye/websocket-driver-node"
 , "author"          : "James Coglan <jcoglan@gmail.com> (http://jcoglan.com/)"
 , "keywords"        : ["websocket"]
+, "license"         : "MIT"
 
-, "version"         : "0.1.0"
+, "version"         : "0.3.4"
 , "engines"         : {"node": ">=0.4.0"}
 , "main"            : "./lib/websocket/driver"
-, "devDependencies" : {"jsclass": ""}
+, "devDependencies" : {"jstest": ""}
 
-, "scripts"         : {"test": "node spec/runner.js"}
+, "scripts"         : {"test": "jstest spec/runner.js"}
+
+, "repository"      : { "type"  : "git"
+                      , "url"   : "git://github.com/faye/websocket-driver-node.git"
+                      }
 
 , "bugs"            : "http://github.com/faye/websocket-driver-node/issues"
-
-, "licenses"        : [ { "type"  : "MIT"
-                        , "url"   : "http://www.opensource.org/licenses/mit-license.php"
-                        }
-                      ]
-
-, "repositories"    : [ { "type"  : "git"
-                        , "url"   : "git://github.com/faye/websocket-driver-node.git"
-                        }
-                      ]
 }
 

spec/runner.js

@@ -1,6 +1,5 @@
-require('jsclass')
-
-var Stream = require('stream').Stream,
+var test   = require('jstest').Test,
+    Stream = require('stream').Stream,
     util   = require('util')
 
 var BufferMatcher = function(data) {
@@ -29,21 +28,18 @@ Collector.prototype.write = function(buffer) {
   return true
 }
 
-JS.require('JS.Test', function() {
-  JS.Test.Unit.TestCase.include({
-    buffer: function(data) {
-      return new BufferMatcher(data)
-    },
-    collector: function() {
-      return this._collector = this._collector || new Collector()
-    }
-  })
-
-  require('./websocket/driver/draft75_examples')
-  require('./websocket/driver/draft75_spec')
-  require('./websocket/driver/draft76_spec')
-  require('./websocket/driver/hybi_spec')
-  require('./websocket/driver/client_spec')
-  JS.Test.autorun()
+test.Unit.TestCase.include({
+  buffer: function(data) {
+    return new BufferMatcher(data)
+  },
+  collector: function() {
+    return this._collector = this._collector || new Collector()
+  }
 })
 
+require('./websocket/driver/draft75_examples')
+require('./websocket/driver/draft75_spec')
+require('./websocket/driver/draft76_spec')
+require('./websocket/driver/hybi_spec')
+require('./websocket/driver/client_spec')
+

spec/websocket/driver/client_spec.js

@@ -1,6 +1,7 @@
-var Client = require("../../../lib/websocket/driver/client")
+var Client = require("../../../lib/websocket/driver/client"),
+    test   = require('jstest').Test
 
-JS.Test.describe("Client", function() { with(this) {
+test.describe("Client", function() { with(this) {
     define("options", function() {
     return this._options = this._options || {protocols: this.protocols()}
   })
@@ -9,9 +10,13 @@ JS.Test.describe("Client", function() { with(this) {
     null
   })
 
+  define("url", function() {
+    return "ws://www.example.com/socket"
+  })
+
   define("driver", function() {
     if (this._driver) return this._driver
-    this._driver = new Client("ws://www.example.com/socket", this.options())
+    this._driver = new Client(this.url(), this.options())
     var self = this
     this._driver.on('open',    function(e) { self.open = true })
     this._driver.on('message', function(e) { self.message += e.data })
@@ -79,6 +84,42 @@ JS.Test.describe("Client", function() { with(this) {
         }})
       }})
 
+      describe("with basic auth", function() { with(this) {
+        define("url", function() { return "ws://user:pass@www.example.com/socket" })
+
+        it("writes the handshake with Authorization", function() { with(this) {
+          expect(driver().io, "emit").given("data", buffer(
+              "GET /socket HTTP/1.1\r\n" +
+              "Host: www.example.com\r\n" +
+              "Upgrade: websocket\r\n" +
+              "Connection: Upgrade\r\n" +
+              "Sec-WebSocket-Key: 2vBVWg4Qyk3ZoM/5d3QD9Q==\r\n" +
+              "Sec-WebSocket-Version: 13\r\n" +
+              "Authorization: Basic dXNlcjpwYXNz\r\n" +
+              "\r\n"))
+          driver().start()
+        }})
+      }})
+
+      describe("with custom headers", function() { with(this) {
+        before(function() { with(this) {
+          driver().setHeader("User-Agent", "Chrome")
+        }})
+
+        it("writes the handshake with custom headers", function() { with(this) {
+          expect(driver().io, "emit").given("data", buffer(
+              "GET /socket HTTP/1.1\r\n" +
+              "Host: www.example.com\r\n" +
+              "Upgrade: websocket\r\n" +
+              "Connection: Upgrade\r\n" +
+              "Sec-WebSocket-Key: 2vBVWg4Qyk3ZoM/5d3QD9Q==\r\n" +
+              "Sec-WebSocket-Version: 13\r\n" +
+              "User-Agent: Chrome\r\n" +
+              "\r\n"))
+          driver().start()
+        }})
+      }})
+
       it("changes the state to connecting", function() { with(this) {
         driver().start()
         assertEqual( "connecting", driver().getState() )
@@ -90,21 +131,33 @@ JS.Test.describe("Client", function() { with(this) {
     before(function() { this.driver().start() })
 
     describe("with a valid response", function() { with(this) {
-      before(function() { this.driver().parse(new Buffer(this.response())) })
+      before(function(resume) { with(this) {
+        driver().parse(new Buffer(response()))
+        setTimeout(resume, 10)
+      }})
 
       it("changes the state to open", function() { with(this) {
         assertEqual( true, open )
         assertEqual( false, close )
         assertEqual( "open", driver().getState() )
       }})
+
+      it("makes the response status available", function() { with(this) {
+        assertEqual( 101, driver().statusCode )
+      }})
+
+      it("makes the response headers available", function() { with(this) {
+        assertEqual( "websocket", driver().headers.upgrade )
+      }})
     }})
 
     describe("with a valid response followed by a frame", function() { with(this) {
-      before(function() { with(this) {
+      before(function(resume) { with(this) {
         var resp = new Buffer(response().length + 4)
         new Buffer(response()).copy(resp)
         new Buffer([0x81, 0x02, 72, 105]).copy(resp, resp.length - 4)
         driver().parse(resp)
+        setTimeout(resume, 10)
       }})
 
       it("changes the state to open", function() { with(this) {
@@ -118,11 +171,27 @@ JS.Test.describe("Client", function() { with(this) {
       }})
     }})
 
+    describe("with a bad status line", function() { with(this) {
+      before(function(resume) { with(this) {
+        var resp = response().replace(/101/g, "4")
+        driver().parse(new Buffer(resp))
+        setTimeout(resume, 10)
+      }})
+
+      it("changes the state to closed", function() { with(this) {
+        assertEqual( false, open )
+        assertEqual( "Error during WebSocket handshake: Unexpected response code: 4", error.message )
+        assertEqual( [1002, "Error during WebSocket handshake: Unexpected response code: 4"], close )
+        assertEqual( "closed", driver().getState() )
+      }})
+    }})
+
     describe("with a bad Upgrade header", function() { with(this) {
-      before(function() {
-        var resp = this.response().replace(/websocket/g, "wrong")
-        this.driver().parse(new Buffer(resp))
-      })
+      before(function(resume) { with(this) {
+        var resp = response().replace(/websocket/g, "wrong")
+        driver().parse(new Buffer(resp))
+        setTimeout(resume, 10)
+      }})
 
       it("changes the state to closed", function() { with(this) {
         assertEqual( false, open )
@@ -133,10 +202,11 @@ JS.Test.describe("Client", function() { with(this) {
     }})
 
     describe("with a bad Accept header", function() { with(this) {
-      before(function() {
-        var resp = this.response().replace(/QV3/g, "wrong")
-        this.driver().parse(new Buffer(resp))
-      })
+      before(function(resume) { with(this) {
+        var resp = response().replace(/QV3/g, "wrong")
+        driver().parse(new Buffer(resp))
+        setTimeout(resume, 10)
+      }})
 
       it("changes the state to closed", function() { with(this) {
         assertEqual( false, open )
@@ -149,10 +219,11 @@ JS.Test.describe("Client", function() { with(this) {
     describe("with valid subprotocols", function() { with(this) {
       define("protocols", function() { return ["foo", "xmpp"] })
 
-      before(function() {
-        var resp = this.response().replace(/\r\n\r\n/, "\r\nSec-WebSocket-Protocol: xmpp\r\n\r\n")
-        this.driver().parse(new Buffer(resp))
-      })
+      before(function(resume) { with(this) {
+        var resp = response().replace(/\r\n\r\n/, "\r\nSec-WebSocket-Protocol: xmpp\r\n\r\n")
+        driver().parse(new Buffer(resp))
+        setTimeout(resume, 10)
+      }})
 
       it("changs the state to open", function() { with(this) {
         assertEqual( true, open )
@@ -168,10 +239,11 @@ JS.Test.describe("Client", function() { with(this) {
     describe("with invalid subprotocols", function() { with(this) {
       define("protocols", function() { return ["foo", "xmpp"] })
 
-      before(function() {
-        var resp = this.response().replace(/\r\n\r\n/, "\r\nSec-WebSocket-Protocol: irc\r\n\r\n")
-        this.driver().parse(new Buffer(resp))
-      })
+      before(function(resume) { with(this) {
+        var resp = response().replace(/\r\n\r\n/, "\r\nSec-WebSocket-Protocol: irc\r\n\r\n")
+        driver().parse(new Buffer(resp))
+        setTimeout(resume, 10)
+      }})
 
       it("changs the state to closed", function() { with(this) {
         assertEqual( false, open )

spec/websocket/driver/draft75_examples.js

@@ -1,4 +1,6 @@
-JS.Test.describe("draft-75", function() { with(this) {
+var test = require('jstest').Test
+
+test.describe("draft-75", function() { with(this) {
   sharedExamplesFor("draft-75 protocol", function() { with(this) {
     describe("in the open state", function() { with(this) {
       before(function() { this.driver().start() })

spec/websocket/driver/draft75_spec.js

@@ -1,6 +1,7 @@
-var Draft75 = require("../../../lib/websocket/driver/draft75")
+var Draft75 = require("../../../lib/websocket/driver/draft75"),
+    test    = require('jstest').Test
 
-JS.Test.describe("Draft75", function() { with(this) {
+test.describe("Draft75", function() { with(this) {
   define("request", function() {
     return this._request = this._request || {
       headers: {

spec/websocket/driver/draft76_spec.js

@@ -1,6 +1,7 @@
-var Draft76 = require("../../../lib/websocket/driver/draft76")
+var Draft76 = require("../../../lib/websocket/driver/draft76"),
+    test    = require('jstest').Test
 
-JS.Test.describe("Draft76", function() { with(this) {
+test.describe("Draft76", function() { with(this) {
   BODY = new Buffer([0x91, 0x25, 0x3e, 0xd3, 0xa9, 0xe7, 0x6a, 0x88])
 
   define("body", function() {
@@ -52,7 +53,7 @@ JS.Test.describe("Draft76", function() { with(this) {
     describe("start", function() { with(this) {
       it("writes the handshake response to the socket", function() { with(this) {
         expect(driver().io, "emit").given("data", buffer(
-            "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" +
+            "HTTP/1.1 101 WebSocket Protocol Handshake\r\n" +
             "Upgrade: WebSocket\r\n" +
             "Connection: Upgrade\r\n" +
             "Sec-WebSocket-Origin: http://www.example.com\r\n" +
@@ -94,7 +95,7 @@ JS.Test.describe("Draft76", function() { with(this) {
 
       it("queues the frames until the handshake has been sent", function() { with(this) {
         expect(driver().io, "emit").given("data", buffer(
-            "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" +
+            "HTTP/1.1 101 WebSocket Protocol Handshake\r\n" +
             "Upgrade: WebSocket\r\n" +
             "Connection: Upgrade\r\n" +
             "Sec-WebSocket-Origin: http://www.example.com\r\n" +
@@ -115,7 +116,7 @@ JS.Test.describe("Draft76", function() { with(this) {
 
       it("writes the handshake response with no body", function() { with(this) {
         expect(driver().io, "emit").given("data", buffer(
-            "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" +
+            "HTTP/1.1 101 WebSocket Protocol Handshake\r\n" +
             "Upgrade: WebSocket\r\n" +
             "Connection: Upgrade\r\n" +
             "Sec-WebSocket-Origin: http://www.example.com\r\n" +

spec/websocket/driver/hybi_spec.js

@@ -1,6 +1,7 @@
-var Hybi = require("../../../lib/websocket/driver/hybi")
+var Hybi = require("../../../lib/websocket/driver/hybi"),
+    test = require('jstest').Test
 
-JS.Test.describe("Hybi", function() { with(this) {
+test.describe("Hybi", function() { with(this) {
   define("request", function() {
     return this._request = this._request || {
       headers: {
@@ -78,6 +79,23 @@ JS.Test.describe("Hybi", function() { with(this) {
         }})
       }})
 
+      describe("with custom headers", function() { with(this) {
+        before(function() { with(this) {
+          driver().setHeader("Authorization", "Bearer WAT")
+        }})
+
+        it("writes the handshake with Sec-WebSocket-Protocol", function() { with(this) {
+          expect(driver().io, "emit").given("data", buffer(
+              "HTTP/1.1 101 Switching Protocols\r\n" +
+              "Upgrade: websocket\r\n" +
+              "Connection: Upgrade\r\n" +
+              "Sec-WebSocket-Accept: JdiiuafpBKRqD7eol0y4vJDTsTs=\r\n" +
+              "Authorization: Bearer WAT\r\n" +
+              "\r\n"))
+          driver().start()
+        }})
+      }})
+
       it("triggers the onopen event", function() { with(this) {
         driver().start()
         assertEqual( true, open )
@@ -181,42 +199,54 @@ JS.Test.describe("Hybi", function() { with(this) {
         return output
       })
 
-      it("parses unmasked text frames", function() { with(this) {
+      it("parses unmasked text frames", function(resume) { with(this) {
         driver().parse([0x81, 0x05, 0x48, 0x65, 0x6c, 0x6c, 0x6f])
-        assertEqual( "Hello", message )
+        setTimeout(function() {
+          resume(function() { assertEqual( "Hello", message ) })
+        }, 10)
       }})
 
-      it("parses multiple frames from the same packet", function() { with(this) {
+      it("parses multiple frames from the same packet", function(resume) { with(this) {
         driver().parse([0x81, 0x05, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x81, 0x05, 0x48, 0x65, 0x6c, 0x6c, 0x6f])
-        assertEqual( "HelloHello", message )
+        setTimeout(function() {
+          resume(function() { assertEqual( "HelloHello", message ) })
+        }, 10)
       }})
 
-      it("parses empty text frames", function() { with(this) {
+      it("parses empty text frames", function(resume) { with(this) {
         driver().parse([0x81, 0x00])
-        assertEqual( "", message )
+        setTimeout(function() {
+          resume(function() { assertEqual( "", message ) })
+        }, 10)
       }})
 
-      it("parses fragmented text frames", function() { with(this) {
+      it("parses fragmented text frames", function(resume) { with(this) {
         driver().parse([0x01, 0x03, 0x48, 0x65, 0x6c])
         driver().parse([0x80, 0x02, 0x6c, 0x6f])
-        assertEqual( "Hello", message )
+        setTimeout(function() {
+          resume(function() { assertEqual( "Hello", message ) })
+        }, 10)
       }})
 
-      it("parses masked text frames", function() { with(this) {
+      it("parses masked text frames", function(resume) { with(this) {
         driver().parse([0x81, 0x85])
         driver().parse(mask())
         driver().parse(maskMessage([0x48, 0x65, 0x6c, 0x6c, 0x6f]))
-        assertEqual( "Hello", message )
+        setTimeout(function() {
+          resume(function() { assertEqual( "Hello", message ) })
+        }, 10)
       }})
 
-      it("parses masked empty text frames", function() { with(this) {
+      it("parses masked empty text frames", function(resume) { with(this) {
         driver().parse([0x81, 0x80])
         driver().parse(mask())
         driver().parse(maskMessage([]))
-        assertEqual( "", message )
+        setTimeout(function() {
+          resume(function() { assertEqual( "", message ) })
+        }, 10)
       }})
 
-      it("parses masked fragmented text frames", function() { with(this) {
+      it("parses masked fragmented text frames", function(resume) { with(this) {
         driver().parse([0x01, 0x81])
         driver().parse(mask())
         driver().parse(maskMessage([0x48]))
@@ -225,49 +255,67 @@ JS.Test.describe("Hybi", function() { with(this) {
         driver().parse(mask())
         driver().parse(maskMessage([0x65, 0x6c, 0x6c, 0x6f]))
 
-        assertEqual( "Hello", message )
+        setTimeout(function() {
+          resume(function() { assertEqual( "Hello", message ) })
+        }, 10)
       }})
 
-      it("closes the socket if the frame has an unrecognized opcode", function() { with(this) {
+      it("closes the socket if the frame has an unrecognized opcode", function(resume) { with(this) {
         driver().parse([0x83, 0x00])
-        assertEqual( [0x88, 0x1e, 0x03, 0xea], collector().bytes.slice(0,4) )
-        assertEqual( "Unrecognized frame opcode: 3", error.message )
-        assertEqual( [1002, "Unrecognized frame opcode: 3"], close )
-        assertEqual( "closed", driver().getState() )
+        setTimeout(function() {
+          resume(function() {
+            assertEqual( [0x88, 0x1e, 0x03, 0xea], collector().bytes.slice(0,4) )
+            assertEqual( "Unrecognized frame opcode: 3", error.message )
+            assertEqual( [1002, "Unrecognized frame opcode: 3"], close )
+            assertEqual( "closed", driver().getState() )
+          })
+        }, 10)
       }})
 
-      it("closes the socket if a close frame is received", function() { with(this) {
+      it("closes the socket if a close frame is received", function(resume) { with(this) {
         driver().parse([0x88, 0x07, 0x03, 0xe8, 0x48, 0x65, 0x6c, 0x6c, 0x6f])
-        assertEqual( [0x88, 0x07, 0x03, 0xe8, 0x48, 0x65, 0x6c, 0x6c, 0x6f], collector().bytes )
-        assertEqual( [1000, "Hello"], close )
-        assertEqual( "closed", driver().getState() )
+        setTimeout(function() {
+          resume(function() {
+            assertEqual( [0x88, 0x07, 0x03, 0xe8, 0x48, 0x65, 0x6c, 0x6c, 0x6f], collector().bytes )
+            assertEqual( [1000, "Hello"], close )
+            assertEqual( "closed", driver().getState() )
+          })
+        }, 10)
       }})
 
-      it("parses unmasked multibyte text frames", function() { with(this) {
+      it("parses unmasked multibyte text frames", function(resume) { with(this) {
         driver().parse([0x81, 0x0b, 0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3, 0xbf])
-        assertEqual( "Apple = ", message )
+        setTimeout(function() {
+          resume(function() { assertEqual( "Apple = ", message ) })
+        }, 10)
       }})
 
-      it("parses frames received in several packets", function() { with(this) {
+      it("parses frames received in several packets", function(resume) { with(this) {
         driver().parse([0x81, 0x0b, 0x41, 0x70, 0x70, 0x6c])
         driver().parse([0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3, 0xbf])
-        assertEqual( "Apple = ", message )
+        setTimeout(function() {
+          resume(function() { assertEqual( "Apple = ", message ) })
+        }, 10)
       }})
 
-      it("parses fragmented multibyte text frames", function() { with(this) {
+      it("parses fragmented multibyte text frames", function(resume) { with(this) {
         driver().parse([0x01, 0x0a, 0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3])
         driver().parse([0x80, 0x01, 0xbf])
-        assertEqual( "Apple = ", message )
+        setTimeout(function() {
+          resume(function() { assertEqual( "Apple = ", message ) })
+        }, 10)
       }})
 
-      it("parse masked multibyte text frames", function() { with(this) {
+      it("parse masked multibyte text frames", function(resume) { with(this) {
         driver().parse([0x81, 0x8b])
         driver().parse(mask())
         driver().parse(maskMessage([0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3, 0xbf]))
-        assertEqual( "Apple = ", message )
+        setTimeout(function() {
+          resume(function() { assertEqual( "Apple = ", message ) })
+        }, 10)
       }})
 
-      it("parses masked fragmented multibyte text frames", function() { with(this) {
+      it("parses masked fragmented multibyte text frames", function(resume) { with(this) {
         driver().parse([0x01, 0x8a])
         driver().parse(mask())
         driver().parse(maskMessage([0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3]))
@@ -276,27 +324,35 @@ JS.Test.describe("Hybi", function() { with(this) {
         driver().parse(mask())
         driver().parse(maskMessage([0xbf]))
 
-        assertEqual( "Apple = ", message )
+        setTimeout(function() {
+          resume(function() { assertEqual( "Apple = ", message ) })
+        }, 10)
       }})
 
-      it("parses unmasked medium-length text frames", function() { with(this) {
+      it("parses unmasked medium-length text frames", function(resume) { with(this) {
         driver().parse([0x81, 0x7e, 0x00, 0xc8])
         var i = 40, result = ""
         while (i--) {
           driver().parse([0x48, 0x65, 0x6c, 0x6c, 0x6f])
           result += "Hello"
         }
-        assertEqual( result, message )
+        setTimeout(function() {
+          resume(function() { assertEqual( result, message ) })
+        }, 10)
       }})
 
-      it("returns an error for too-large frames", function() { with(this) {
-        driver().parse([0x81, 0x7f, 0x00, 0x20, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00])
-        assertEqual( "WebSocket frame length too large", error.message )
-        assertEqual( [1009, "WebSocket frame length too large"], close )
-        assertEqual( "closed", driver().getState() )
+      it("returns an error for too-large frames", function(resume) { with(this) {
+        driver().parse([0x81, 0x7f, 0x00, 0x00, 0x00, 0x00, 0x40, 0x00, 0x00, 0x00])
+        setTimeout(function() {
+          resume(function() {
+            assertEqual( "WebSocket frame length too large", error.message )
+            assertEqual( [1009, "WebSocket frame length too large"], close )
+            assertEqual( "closed", driver().getState() )
+          })
+        }, 10)
       }})
 
-      it("parses masked medium-length text frames", function() { with(this) {
+      it("parses masked medium-length text frames", function(resume) { with(this) {
         driver().parse([0x81, 0xfe, 0x00, 0xc8])
         driver().parse(mask())
         var i = 40, result = "", packet = []
@@ -305,12 +361,18 @@ JS.Test.describe("Hybi", function() { with(this) {
           result += "Hello"
         }
         driver().parse(maskMessage(packet))
-        assertEqual( result, message )
+        setTimeout(function() {
+          resume(function() { assertEqual( result, message ) })
+        }, 10)
       }})
 
-      it("replies to pings with a pong", function() { with(this) {
+      it("replies to pings with a pong", function(resume) { with(this) {
         driver().parse([0x89, 0x04, 0x4f, 0x48, 0x41, 0x49])
-        assertEqual( [0x8a, 0x04, 0x4f, 0x48, 0x41, 0x49], collector().bytes )
+        setTimeout(function() {
+          resume(function() {
+            assertEqual( [0x8a, 0x04, 0x4f, 0x48, 0x41, 0x49], collector().bytes )
+          })
+        }, 10)
       }})
     }})
 
@@ -361,18 +423,22 @@ JS.Test.describe("Hybi", function() { with(this) {
         assertEqual( true, driver().ping() )
       }})
 
-      it("runs the given callback on mathing pong", function() { with(this) {
+      it("runs the given callback on mathing pong", function(resume) { with(this) {
         var reply = null
         driver().ping("Hi", function() { reply = true })
         driver().parse([0x8a, 0x02, 72, 105])
-        assert( reply )
+        setTimeout(function() {
+          resume(function() { assert( reply ) })
+        }, 10)
       }})
 
-      it("does not run the callback on non-matching pong", function() { with(this) {
+      it("does not run the callback on non-matching pong", function(resume) { with(this) {
         var reply = null
         driver().ping("Hi", function() { reply = true })
         driver().parse([0x8a, 0x03, 119, 97, 116])
-        assert( !reply )
+        setTimeout(function() {
+          resume(function() { assert( !reply ) })
+        }, 10)
       }})
     }})
 
@@ -409,15 +475,21 @@ JS.Test.describe("Hybi", function() { with(this) {
       this.driver().start()
     })
 
-    it("does not emit a message", function() { with(this) {
+    it("does not emit a message", function(resume) { with(this) {
       driver().parse([0x81, 0x05, 0x48, 0x65, 0x6c, 0x6c, 0x6f])
-      assertEqual( "", message )
+      setTimeout(function() {
+        resume(function() { assertEqual( "", message ) })
+      }, 10)
     }})
 
-    it("returns an error", function() { with(this) {
+    it("returns an error", function(resume) { with(this) {
       driver().parse([0x81, 0x05, 0x48, 0x65, 0x6c, 0x6c, 0x6f])
-      assertEqual( "Received unmasked frame but masking is required", error.message )
-      assertEqual( [1003, "Received unmasked frame but masking is required"], close )
+      setTimeout(function() {
+        resume(function() {
+          assertEqual( "Received unmasked frame but masking is required", error.message )
+          assertEqual( [1003, "Received unmasked frame but masking is required"], close )
+        })
+      }, 10)
     }})
   }})
 
@@ -461,8 +533,9 @@ JS.Test.describe("Hybi", function() { with(this) {
     }})
 
     describe("receiving a close frame", function() { with(this) {
-      before(function() {
+      before(function(resume) {
         this.driver().parse([0x88, 0x04, 0x03, 0xe9, 0x4f, 0x4b])
+        setTimeout(resume, 10)
       })
 
       it("triggers the onclose event", function() { with(this) {
@@ -476,10 +549,11 @@ JS.Test.describe("Hybi", function() { with(this) {
   }})
 
   describe("in the closed state", function() { with(this) {
-    before(function() {
+    before(function(resume) {
       this.driver().start()
       this.driver().close()
       this.driver().parse([0x88, 0x02, 0x03, 0xea])
+      setTimeout(resume, 10)
     })
 
     describe("frame", function() { with(this) {