[WIP] Begin implementing permessage-deflate.

Implement masking as a transform stream.
Optimise some special cases in StreamReader; return a new Buffer from read() immediately if there's enough buffered data, and if the read can be fulfilled entirely from the first queued buffer than take a slice rather than copying into a new buffer.
2014-05-11 22:32:15 +01:00 · 2014-05-11 22:07:34 +01:00 · 2014-05-11 13:33:51 +01:00 · 2014-05-09 15:35:40 +01:00 · 2014-05-08 16:35:21 +01:00
26 changed files with 584 additions and 526 deletions
@@ -8,3 +8,4 @@ node_js:

 before_install:
  - '[ "${TRAVIS_NODE_VERSION}" = "0.6" ] && npm conf set strict-ssl false || true'
+
@@ -1,16 +1,3 @@
-### 0.4.0 / 2014-11-08
-
-* Support connection via HTTP proxies using `CONNECT`
-
-### 0.3.6 / 2014-10-04
-
-* It is now possible to call `close()` before `start()` and close the driver
-
-### 0.3.5 / 2014-07-06
-
-* Don't hold references to frame buffers after a message has been emitted
-* Make sure that `protocol` and `version` are exposed properly by the TCP driver
-
 ### 0.3.4 / 2014-05-08

 * Don't hold memory-leaking references to I/O buffers after they have been parsed
@@ -52,3 +39,4 @@
 ### 0.1.0 / 2013-05-04

 * First stable release
+
@@ -1,4 +1,4 @@
-# websocket-driver [![Build Status](https://travis-ci.org/faye/websocket-driver-node.svg)](https://travis-ci.org/faye/websocket-driver-node)
+# websocket-driver [![Build Status](https://travis-ci.org/faye/websocket-driver-node.png)](https://travis-ci.org/faye/websocket-driver-node)

 This module provides a complete implementation of the WebSocket protocols that
 can be hooked up to any I/O stream. It aims to simplify things by decoupling
@@ -134,17 +134,17 @@ var net = require('net'),
    websocket = require('websocket-driver');

 var driver = websocket.client('ws://www.example.com/socket'),
-    tcp = net.connect(80, 'www.example.com');
+    tcp = net.createConnection(80, 'www.example.com');

 tcp.pipe(driver.io).pipe(tcp);

-tcp.on('connect', function() {
-  driver.start();
-});
-
 driver.messages.on('data', function(message) {
  console.log('Got a message', message);
 });
+
+tcp.on('connect', function() {
+  driver.start();
+});
 ```

 Client drivers have two additional properties for reading the HTTP data that
@@ -154,65 +154,6 @@ was sent back by the server:
 * `driver.headers` - an object containing the response headers


-### HTTP Proxies
-
-The client driver supports connections via HTTP proxies using the `CONNECT`
-method. Instead of sending the WebSocket handshake immediately, it will send a
-`CONNECT` request, wait for a `200` response, and then proceed as normal.
-
-To use this feature, call `driver.proxy(url)` where `url` is the origin of the
-proxy, including a username and password if required. This produces a duplex
-stream that you should pipe in and out of your TCP connection to the proxy
-server. When the proxy emits `connect`, you can then pipe `driver.io` to your
-TCP stream and call `driver.start()`.
-
-```js
-var net = require('net'),
-    websocket = require('websocket-driver');
-
-var driver = websocket.client('ws://www.example.com/socket'),
-    proxy  = driver.proxy('http://username:password@proxy.example.com'),
-    tcp    = net.connect(80, 'proxy.example.com');
-
-tcp.pipe(proxy).pipe(tcp, {end: false});
-
-tcp.on('connect', function() {
-  proxy.start();
-});
-
-proxy.on('connect', function() {
-  driver.io.pipe(tcp).pipe(driver.io);
-  driver.start();
-});
-
-driver.messages.on('data', function(message) {
-  console.log('Got a message', message);
-});
-```
-
-The proxy's `connect` event is also where you should perform a TLS handshake on
-your TCP stream, if you are connecting to a `wss:` endpoint.
-
-In the event that proxy connection fails, `proxy` will emit an `error`. You can
-inspect the proxy's response via `proxy.statusCode` and `proxy.headers`.
-
-```js
-proxy.on('error', function(error) {
-  console.error(error.message);
-  console.log(proxy.statusCode);
-  console.log(proxy.headers);
-});
-```
-
-Before calling `proxy.start()` you can set custom headers using
-`proxy.setHeader()`:
-
-```js
-proxy.setHeader('User-Agent', 'node');
-proxy.start();
-```
-
-
 ### Driver API

 Drivers are created using one of the following methods:
@@ -363,3 +304,4 @@ AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 SOFTWARE.
+
@@ -1,8 +1,10 @@
 var net       = require('net'),
-    websocket = require('../lib/websocket/driver');
+    websocket = require('../lib/websocket/driver'),
+    deflate   = require('../lib/websocket/extensions/per_message_deflate');

 var server = net.createServer(function(connection) {
  var driver = websocket.server();
+  driver.addExtension('permessage-deflate', deflate.create());

  driver.on('connect', function() {
    if (websocket.isWebSocket(driver)) driver.start();
@@ -18,3 +20,4 @@ var server = net.createServer(function(connection) {
 });

 server.listen(process.argv[2]);
+
@@ -41,3 +41,4 @@ var Driver = {
 };

 module.exports = Driver;
+
@@ -6,13 +6,14 @@ var Emitter = require('events').EventEmitter,
 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;
+  this._request    = request;
+  this._options    = options || {};
+  this._extensions = {};
+  this._maxLength  = this._options.maxLength || this.MAX_LENGTH;
+  this.__headers   = new Headers();
+  this.__queue     = [];
+  this.readyState  = 0;
+  this.url         = url;

  this.io = new streams.IO(this);
  this.messages = new streams.Messages(this);
@@ -51,13 +52,17 @@ var instance = {
    });
  },

+  addExtension: function(name, extension) {
+    this._extensions[name] = extension;
+  },
+
  getState: function() {
    return this.STATES[this.readyState] || null;
  },

  setHeader: function(name, value) {
    if (this.readyState > 0) return false;
-    this._headers.set(name, value);
+    this.__headers.set(name, value);
    return true;
  },

@@ -108,6 +113,9 @@ var instance = {
 for (var key in instance)
  Base.prototype[key] = instance[key];

+Base.parseHeader = function(header) {
+  return Headers.parseHeader(header);
+};

 Base.ConnectEvent = function() {};

@@ -123,3 +131,4 @@ Base.MessageEvent = function(data) {
 };

 module.exports = Base;
+
@@ -1,49 +1,27 @@
 var url        = require('url'),
    util       = require('util'),
-    HttpParser = require('../http_parser'),
+    HttpParser = require('./http_parser'),
    Base       = require('./base'),
-    Hybi       = require('./hybi'),
-    Proxy      = require('./proxy');
+    Hybi       = require('./hybi');

-var Client = function(_url, options) {
+var Client = function(url, options) {
  this.version = 'hybi-13';
-  Hybi.call(this, null, _url, options);
+  Hybi.call(this, null, url, options);

  this.readyState = -1;
  this._key       = Client.generateKey();
  this._accept    = Hybi.generateAccept(this._key);
  this._http      = new HttpParser('response');
-
-  var uri  = url.parse(this.url),
-      auth = uri.auth && new Buffer(uri.auth, 'utf8').toString('base64');
-
-  this._pathname = (uri.pathname || '/') + (uri.search || '');
-
-  this._headers.set('Host', uri.host);
-  this._headers.set('Upgrade', 'websocket');
-  this._headers.set('Connection', 'Upgrade');
-  this._headers.set('Sec-WebSocket-Key', this._key);
-  this._headers.set('Sec-WebSocket-Version', '13');
-
-  if (this._protocols.length > 0)
-    this._headers.set('Sec-WebSocket-Protocol', this._protocols.join(', '));
-
-  if (auth)
-    this._headers.set('Authorization', 'Basic ' + auth);
 };
 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');
 };

 var instance = {
-  proxy: function(origin, options) {
-    return new Proxy(this, origin, options);
-  },
-
  start: function() {
    if (this.readyState !== -1) return false;
    this._write(this._handshakeRequest());
@@ -56,16 +34,29 @@ var instance = {

    this._http.parse(data);
    if (!this._http.isComplete()) return;
-
+    
    this._validateHandshake();
    this.parse(this._http.body);
  },

  _handshakeRequest: function() {
-    var start   = 'GET ' + this._pathname + ' HTTP/1.1',
-        headers = [start, this._headers.toString(), ''];
+    var uri = url.parse(this.url);

-    return new Buffer(headers.join('\r\n'), 'utf8');
+    var headers = [ 'GET ' + (uri.pathname || '/') + (uri.search || '') + ' HTTP/1.1',
+                    'Host: ' + uri.hostname + (uri.port ? ':' + uri.port : ''),
+                    'Upgrade: websocket',
+                    'Connection: Upgrade',
+                    'Sec-WebSocket-Key: ' + this._key,
+                    'Sec-WebSocket-Version: 13'
+                  ];
+
+    if (this._protocols.length > 0)
+      headers.push('Sec-WebSocket-Protocol: ' + this._protocols.join(', '));
+
+    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) {
@@ -118,3 +109,4 @@ for (var key in instance)
  Client.prototype[key] = instance[key];

 module.exports = Client;
+
@@ -5,11 +5,6 @@ var Draft75 = function(request, url, options) {
  Base.apply(this, arguments);
  this._stage = 0;
  this.version = 'hixie-75';
-
-  this._headers.set('Upgrade', 'WebSocket');
-  this._headers.set('Connection', 'Upgrade');
-  this._headers.set('WebSocket-Origin', this._request.headers.origin);
-  this._headers.set('WebSocket-Location', this.url);
 };
 util.inherits(Draft75, Base);

@@ -93,10 +88,14 @@ var instance = {
  },

  _handshakeResponse: function() {
-    var start   = 'HTTP/1.1 101 Web Socket Protocol Handshake',
-        headers = [start, this._headers.toString(), ''];
-
-    return new Buffer(headers.join('\r\n'), 'utf8');
+    return new Buffer('HTTP/1.1 101 Web Socket Protocol Handshake\r\n' +
+                      'Upgrade: WebSocket\r\n' +
+                      'Connection: Upgrade\r\n' +
+                      'WebSocket-Origin: ' + this._request.headers.origin + '\r\n' +
+                      'WebSocket-Location: ' + this.url + '\r\n' +
+                      this.__headers.toString() +
+                      '\r\n',
+                      'utf8');
  },

  _parseLeadingByte: function(data) {
@@ -116,3 +115,4 @@ for (var key in instance)
  Draft75.prototype[key] = instance[key];

 module.exports = Draft75;
+
@@ -26,13 +26,6 @@ var Draft76 = function(request, url, options) {
  this._stage  = -1;
  this._body   = [];
  this.version = 'hixie-76';
-
-  this._headers.clear();
-
-  this._headers.set('Upgrade', 'WebSocket');
-  this._headers.set('Connection', 'Upgrade');
-  this._headers.set('Sec-WebSocket-Origin', this._request.headers.origin);
-  this._headers.set('Sec-WebSocket-Location', this.url);
 };
 util.inherits(Draft76, Draft75);

@@ -55,10 +48,14 @@ var instance = {
  },

  _handshakeResponse: function() {
-    var start   = 'HTTP/1.1 101 WebSocket Protocol Handshake',
-        headers = [start, this._headers.toString(), ''];
-
-    return new Buffer(headers.join('\r\n'), 'binary');
+    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');
  },

  _handshakeSignature: function() {
@@ -109,3 +106,4 @@ for (var key in instance)
  Draft76.prototype[key] = instance[key];

 module.exports = Draft76;
+
@@ -1,14 +1,26 @@
 var Headers = function() {
-  this.clear();
-};
-
-Headers.prototype.ALLOWED_DUPLICATES = ['set-cookie', 'set-cookie2', 'warning', 'www-authenticate'];
-
-Headers.prototype.clear = function() {
  this._sent  = {};
  this._lines = [];
 };

+Headers.parseHeader = function(header) {
+  if (!header) return [];
+
+  return header.split(/\s*,\s*/).map(function(value) {
+    var parts  = value.split(/\s*;\s*/),
+        name   = parts.shift(),
+        params = {};
+
+    parts.forEach(function(part) {
+      var pair = part.split(/\s*=\s*/);
+      params[pair[0]] = pair[1] || true;
+    });
+    return {name: name, params: params};
+  });
+};
+
+Headers.prototype.ALLOWED_DUPLICATES = ['set-cookie', 'set-cookie2', 'warning', 'www-authenticate']
+
 Headers.prototype.set = function(name, value) {
  if (value === undefined) return;

@@ -16,7 +28,7 @@ Headers.prototype.set = function(name, value) {
  value = this._strip(value);

  var key = name.toLowerCase();
-  if (!this._sent.hasOwnProperty(key) || this.ALLOWED_DUPLICATES.indexOf(key) >= 0) {
+  if (!this._sent.hasOwnProperty(key) || this.ALLOWED_DUPLICATES.indexOf(key) < 0) {
    this._sent[key] = true;
    this._lines.push(name + ': ' + value + '\r\n');
  }
@@ -31,3 +43,4 @@ Headers.prototype._strip = function(string) {
 };

 module.exports = Headers;
+
@@ -19,20 +19,22 @@ var HttpParser = function(type) {
  };

  this._parser.onHeaderValue = function(b, start, length) {
-    self.headers[current] = b.toString('utf8', start, start + length);
+    self.set(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;
+    
+    var headers = info.headers, name, values;
    if (!headers) return;

    for (var i = 0, n = headers.length; i < n; i += 2)
-      self.headers[headers[i].toLowerCase()] = headers[i+1];
+      self.set(headers[i].toLowerCase(), headers[i+1]);
+  };

+  this._parser.onMessageComplete = this._parser[HTTPParser.kOnMessageComplete] = function() {
    self._complete = true;
  };
 };
@@ -61,6 +63,13 @@ HttpParser.METHODS = {
  24: 'PATCH'
 };

+HttpParser.prototype.set = function(name, value) {
+  var values = [];
+  if (this.headers.hasOwnProperty(name)) values.push(this.headers[name]);
+  values.push(value);
+  this.headers[name] = values.join(', ');
+};
+
 HttpParser.prototype.isComplete = function() {
  return this._complete;
 };
@@ -76,3 +85,4 @@ HttpParser.prototype.parse = function(data) {
 };

 module.exports = HttpParser;
+
@@ -1,53 +1,33 @@
-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._masking        = this._options.masking;
-  this._protocols      = this._options.protocols || [];
-  this._requireMasking = this._options.requireMasking;
-  this._pingCallbacks  = {};
+  this._reader    = new Reader({context: this});
+  this._masking   = this._options.masking;
+  this._protocols = this._options.protocols || [];

  if (typeof this._protocols === 'string')
    this._protocols = this._protocols.split(/\s*,\s*/);

-  if (!this._request) return;
+  this._requireMasking = this._options.requireMasking;
+  this._pingCallbacks  = {};

-  var secKey    = this._request.headers['sec-websocket-key'],
-      protos    = this._request.headers['sec-websocket-protocol'],
-      version   = this._request.headers['sec-websocket-version'],
-      supported = this._protocols;
-
-  this._headers.set('Upgrade', 'websocket');
-  this._headers.set('Connection', 'Upgrade');
-  this._headers.set('Sec-WebSocket-Accept', Hybi.generateAccept(secKey));
-
-  if (protos !== undefined) {
-    if (typeof protos === 'string') protos = protos.split(/\s*,\s*/);
-    this.protocol = protos.filter(function(p) { return supported.indexOf(p) >= 0 })[0];
-    if (this.protocol) this._headers.set('Sec-WebSocket-Protocol', this.protocol);
+  if (!this.version) {
+    var version = this._request.headers['sec-websocket-version'];
+    this.version = 'hybi-' + version;
  }

-  this.version = 'hybi-' + version;
+  this._reader.read(1, this._parseOpcode);
 };
 util.inherits(Hybi, Base);

-Hybi.mask = function(payload, mask, offset) {
-  if (!mask || 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);
@@ -101,45 +81,7 @@ 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._emitFrame(buffer);
-            this._stage = 0;
-          }
-          break;
-
-        default:
-          buffer = null;
-      }
-    }
+    this._reader.write(data);
  },

  frame: function(data, type, code) {
@@ -166,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);
@@ -215,7 +157,7 @@ var instance = {
    reason = reason || '';
    code   = code   || this.ERRORS.normal_closure;

-    if (this.readyState <= 0) {
+    if (this.readyState === 0) {
      this.readyState = 3;
      this.emit('close', new Base.CloseEvent(code, reason));
      return true;
@@ -229,16 +171,52 @@ var instance = {
  },

  _handshakeResponse: function() {
-    var start   = 'HTTP/1.1 101 Switching Protocols',
-        headers = [start, this._headers.toString(), ''];
+    var secKey = this._request.headers['sec-websocket-key'];
+    if (!secKey) return '';

-    return new Buffer(headers.join('\r\n'), 'utf8');
+    var accept    = Hybi.generateAccept(secKey),
+        offers    = Base.parseHeader(this._request.headers['sec-websocket-extensions']),
+        protos    = this._request.headers['sec-websocket-protocol'],
+        supported = this._protocols,
+        proto,
+
+        headers = [
+          'HTTP/1.1 101 Switching Protocols',
+          'Upgrade: websocket',
+          'Connection: Upgrade',
+          'Sec-WebSocket-Accept: ' + accept
+        ];
+
+    if (protos !== undefined) {
+      if (typeof protos === 'string') protos = protos.split(/\s*,\s*/);
+      proto = protos.filter(function(p) { return supported.indexOf(p) >= 0 })[0];
+      if (proto) {
+        this.protocol = proto;
+        headers.push('Sec-WebSocket-Protocol: ' + proto);
+      }
+    }
+
+    // TODO: implement a selection procedure for extensions
+    // see http://tools.ietf.org/html/draft-ietf-hybi-permessage-compression-17#section-5
+    var extensions = offers.map(function(offer) {
+      var ext = this._extensions[offer.name];
+      return ext && ext.createSession(offer.params);
+    }, this).filter(function(k) {
+      return k;
+    });
+
+    if (extensions.length > 0) {
+      extensions = extensions.map(function(e) { return e.responseHeader() });
+      headers.push('Sec-WebSocket-Extensions: ' + extensions.join(', '));
+    }
+
+    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._stage = 5;
+    this._reader.end();
    this.emit('close', new Base.CloseEvent(code, reason));
  },

@@ -247,7 +225,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;
    });
@@ -258,8 +238,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._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);
@@ -270,10 +251,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');
@@ -282,10 +265,10 @@ var instance = {

    if (this._length >= 0 && this._length <= 125) {
      if (!this._checkFrameLength()) return;
-      this._stage = this._masked ? 3 : 4;
+      this._readMask();
    } else {
-      this._lengthSize = (this._length === 126 ? 2 : 8);
-      this._stage      = 2;
+      var lengthSize = (this._length === 126 ? 2 : 8);
+      this._reader.read(lengthSize, this._parseExtendedLength);
    }
  },

@@ -297,7 +280,7 @@ var instance = {

    if (!this._checkFrameLength()) return;

-    this._stage  = this._masked ? 3 : 4;
+    this._readMask();
  },

  _checkFrameLength: function() {
@@ -309,18 +292,33 @@ var instance = {
    }
  },

-  _emitFrame: function(buffer) {
-    var payload = Hybi.mask(buffer, this._mask),
-        isFinal = this._final,
-        opcode  = this._opcode;
+  _readMask: function() {
+    if (this._masked)
+      this._reader.read(4, function(buffer) {
+        this._mask = new Mask(buffer);
+        this._readPayload();
+      });
+    else
+      this._readPayload();
+  },

-    this._final = this._opcode = this._length = this._lengthSize = this._masked = this._mask = null;
+  _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 (isFinal) {
-        var message = this._concatBuffer();
+      if (this._final) {
+        var message = Concat.concatBuffers(this.__buffer, this.__blength);
        if (this._mode === 'text') message = this._encode(message);
        this._reset();
        if (message === null)
@@ -330,7 +328,7 @@ var instance = {
      }
    }
    else if (opcode === this.OPCODES.text) {
-      if (isFinal) {
+      if (this._final) {
        var message = this._encode(payload);
        if (message === null)
          this._fail('encoding_error', 'Could not decode a text frame as UTF-8');
@@ -342,7 +340,7 @@ var instance = {
      }
    }
    else if (opcode === this.OPCODES.binary) {
-      if (isFinal) {
+      if (this._final) {
        this.emit('message', new Base.MessageEvent(payload));
      } else {
        this._mode = 'binary';
@@ -374,6 +372,8 @@ var instance = {
      delete callbacks[message];
      if (callback) callback()
    }
+
+    this._reader.read(1, this._parseOpcode);
  },

  _buffer: function(fragment) {
@@ -381,17 +381,6 @@ var instance = {
    this.__blength += fragment.length;
  },

-  _concatBuffer: function() {
-    var buffer = new Buffer(this.__blength),
-        offset = 0;
-
-    for (var i = 0, n = this.__buffer.length; i < n; i++) {
-      this.__buffer[i].copy(buffer, offset);
-      offset += this.__buffer[i].length;
-    }
-    return buffer;
-  },
-
  _reset: function() {
    this._mode     = null;
    this.__buffer  = [];
@@ -418,3 +407,4 @@ for (var key in instance)
  Hybi.prototype[key] = instance[key];

 module.exports = Hybi;
+
@@ -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;
@@ -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;
+
@@ -1,25 +1,121 @@
-var StreamReader = function() {
+var Stream = require('stream').Stream,
+    Concat = require('./concat'),
+    util   = require('util');
+
+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.read = function(length) {
-  if (length > this._queueSize) return null;
+StreamReader.prototype.end = function(buffer) {
+  if (buffer) this.write(buffer);
+  this.writable = false;

-  var buffer = new Buffer(length),
-      queue  = this._queue,
+  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, size;
+      buffer, chunk, size;
+
+  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];
@@ -1,95 +0,0 @@
-var Stream     = require('stream').Stream,
-    url        = require('url'),
-    util       = require('util'),
-    Headers    = require('./headers'),
-    HttpParser = require('../http_parser');
-
-var PORTS = {'ws:': 80, 'wss:': 443};
-
-var Proxy = function(client, origin, options) {
-  this._client  = client;
-  this._http    = new HttpParser('response');
-  this._origin  = (typeof client.url === 'object') ? client.url : url.parse(client.url);
-  this._url     = (typeof origin === 'object') ? origin : url.parse(origin);
-  this._options = options || {};
-  this._state   = 0;
-
-  this.readable = this.writable = true;
-  this._paused  = false;
-
-  this._headers = new Headers();
-  this._headers.set('Host', this._origin.host);
-  this._headers.set('Connection', 'keep-alive');
-  this._headers.set('Proxy-Connection', 'keep-alive');
-
-  var auth = this._url.auth && new Buffer(this._url.auth, 'utf8').toString('base64');
-  if (auth) this._headers.set('Proxy-Authorization', 'Basic ' + auth);
-};
-util.inherits(Proxy, Stream);
-
-var instance = {
-  setHeader: function(name, value) {
-    if (this._state !== 0) return false;
-    this._headers.set(name, value);
-    return true;
-  },
-
-  start: function() {
-    if (this._state !== 0) return false;
-    this._state = 1;
-
-    var origin = this._origin,
-        port   = origin.port || PORTS[origin.protocol],
-        start  = 'CONNECT ' + origin.hostname + ':' + port + ' HTTP/1.1';
-
-    var headers = [start, this._headers.toString(), ''];
-
-    this.emit('data', new Buffer(headers.join('\r\n'), 'utf8'));
-    return true;
-  },
-
-  pause: function() {
-    this._paused = true;
-  },
-
-  resume: function() {
-    this._paused = false;
-    this.emit('drain');
-  },
-
-  write: function(chunk) {
-    if (!this.writable) return false;
-
-    this._http.parse(chunk);
-    if (!this._http.isComplete()) return !this._paused;
-
-    this.statusCode = this._http.statusCode;
-    this.headers    = this._http.headers;
-
-    if (this.statusCode === 200) {
-      this.emit('connect');
-    } else {
-      var message = "Can't establish a connection to the server at " + this._origin.href;
-      this.emit('error', new Error(message));
-    }
-    this.end();
-    return !this._paused;
-  },
-
-  end: function(chunk) {
-    if (!this.writable) return;
-    if (chunk !== undefined) this.write(chunk);
-    this.readable = this.writable = false;
-    this.emit('close');
-    this.emit('end');
-  },
-
-  destroy: function() {
-    this.end();
-  }
-};
-
-for (var key in instance)
-  Proxy.prototype[key] = instance[key];
-
-module.exports = Proxy;
@@ -1,5 +1,5 @@
 var util       = require('util'),
-    HttpParser = require('../http_parser'),
+    HttpParser = require('./http_parser'),
    Base       = require('./base'),
    Draft75    = require('./draft75'),
    Draft76    = require('./draft76'),
@@ -34,15 +34,13 @@ var instance = {
    this._delegate = Server.http(this, this._options);
    this._delegate.messages = this.messages;
    this._delegate.io = this.io;
+    this._delegate._extensions = this._extensions;

    this._delegate.on('open', function() { self._open() });
    this.EVENTS.forEach(function(event) {
      this._delegate.on(event, function(e) { self.emit(event, e) });
    }, this);

-    this.protocol = this._delegate.protocol;
-    this.version  = this._delegate.version;
-
    this.parse(this._http.body);
    this.emit('connect', new Base.ConnectEvent());
  },
@@ -104,3 +102,4 @@ Server.http = function(request, options) {
 };

 module.exports = Server;
+
@@ -0,0 +1,22 @@
+var PerMessageDeflate = function() {
+  this.name = 'permessage-deflate';
+};
+
+PerMessageDeflate.create = function() {
+  return new this();
+};
+
+PerMessageDeflate.prototype.createSession = function(params) {
+  return new Session(this, params);
+};
+
+var Session = function(extension, params) {
+  this._ext    = extension;
+  this._params = params;
+};
+
+Session.prototype.responseHeader = function() {
+  return this._ext.name;
+};
+
+module.exports = PerMessageDeflate;
@@ -141,3 +141,4 @@ Messages.prototype.destroy = function() {};

 exports.IO = IO;
 exports.Messages = Messages;
+
@@ -5,7 +5,7 @@
 , "keywords"        : ["websocket"]
 , "license"         : "MIT"

-, "version"         : "0.4.0"
+, "version"         : "0.3.4"
 , "engines"         : {"node": ">=0.4.0"}
 , "main"            : "./lib/websocket/driver"
 , "devDependencies" : {"jstest": ""}
@@ -18,3 +18,4 @@

 , "bugs"            : "http://github.com/faye/websocket-driver-node/issues"
 }
+
@@ -42,3 +42,4 @@ require('./websocket/driver/draft75_spec')
 require('./websocket/driver/draft76_spec')
 require('./websocket/driver/hybi_spec')
 require('./websocket/driver/client_spec')
+
@@ -50,14 +50,6 @@ test.describe("Client", function() { with(this) {
      assertEqual( null, driver().getState() )
    }})

-    describe("close", function() { with(this) {
-      it("changes the state to closed", function() { with(this) {
-        driver().close()
-        assertEqual( "closed", driver().getState() )
-        assertEqual( [1000, ''], close )
-      }})
-    }})
-
    describe("start", function() { with(this) {
      it("writes the handshake request to the socket", function() { with(this) {
        expect(driver().io, "emit").given("data", buffer(
@@ -135,75 +127,14 @@ test.describe("Client", function() { with(this) {
    }})
  }})

-  describe("using a proxy", function() { with(this) {
-    it("sends a CONNECT request", function() { with(this) {
-      var proxy = driver().proxy("http://proxy.example.com")
-      expect(proxy, "emit").given("data", buffer(
-          "CONNECT www.example.com:80 HTTP/1.1\r\n" +
-          "Host: www.example.com\r\n" +
-          "Connection: keep-alive\r\n" +
-          "Proxy-Connection: keep-alive\r\n" +
-          "\r\n"))
-      proxy.start()
-    }})
-
-    it("sends an authenticated CONNECT request", function() { with(this) {
-      var proxy = driver().proxy("http://user:pass@proxy.example.com")
-      expect(proxy, "emit").given("data", buffer(
-          "CONNECT www.example.com:80 HTTP/1.1\r\n" +
-          "Host: www.example.com\r\n" +
-          "Connection: keep-alive\r\n" +
-          "Proxy-Connection: keep-alive\r\n" +
-          "Proxy-Authorization: Basic dXNlcjpwYXNz\r\n" +
-          "\r\n"))
-      proxy.start()
-    }})
-
-    it("sends a CONNECT request with custom headers", function() { with(this) {
-      var proxy = driver().proxy("http://user:pass@proxy.example.com")
-      proxy.setHeader("User-Agent", "Chrome")
-      expect(proxy, "emit").given("data", buffer(
-          "CONNECT www.example.com:80 HTTP/1.1\r\n" +
-          "Host: www.example.com\r\n" +
-          "Connection: keep-alive\r\n" +
-          "Proxy-Connection: keep-alive\r\n" +
-          "Proxy-Authorization: Basic dXNlcjpwYXNz\r\n" +
-          "User-Agent: Chrome\r\n" +
-          "\r\n"))
-      proxy.start()
-    }})
-
-    describe("receiving a response", function() { with(this) {
-      before(function() { with(this) {
-        this.proxy = driver().proxy("http://proxy.example.com")
-      }})
-
-      it("returns true when the response is written", function() { with(this) {
-        // this prevents downstream connections suddenly closing for no reason
-        assertEqual( true, proxy.write(new Buffer("HTTP/1.1 200 OK\r\n\r\n")) )
-      }})
-
-      it("emits a 'connect' event when the proxy connects", function() { with(this) {
-        expect(proxy, "emit").given("connect")
-        expect(proxy, "emit").given("close")
-        expect(proxy, "emit").given("end")
-        proxy.write(new Buffer("HTTP/1.1 200 OK\r\n\r\n"))
-      }})
-
-      it("emits an 'error' event if the proxy does not connect", function() { with(this) {
-        expect(proxy, "emit").given("error", objectIncluding({message: "Can't establish a connection to the server at ws://www.example.com/socket"}))
-        expect(proxy, "emit").given("close")
-        expect(proxy, "emit").given("end")
-        proxy.write(new Buffer("HTTP/1.1 403 Forbidden\r\n\r\n"))
-      }})
-    }})
-  }})
-
  describe("in the connecting state", 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 )
@@ -221,11 +152,12 @@ test.describe("Client", function() { with(this) {
    }})

    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) {
@@ -240,10 +172,11 @@ test.describe("Client", function() { with(this) {
    }})

    describe("with a bad status line", function() { with(this) {
-      before(function() {
-        var resp = this.response().replace(/101/g, "4")
-        this.driver().parse(new Buffer(resp))
-      })
+      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 )
@@ -254,10 +187,11 @@ test.describe("Client", function() { with(this) {
    }})

    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 )
@@ -268,10 +202,11 @@ 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 )
@@ -284,10 +219,11 @@ 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 )
@@ -303,10 +239,11 @@ 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 )
@@ -321,3 +258,4 @@ test.describe("Client", function() { with(this) {
    }})
  }})
 }})
+
@@ -112,3 +112,4 @@ test.describe("draft-75", function() { with(this) {
    }})
  }})
 }})
+
@@ -97,3 +97,4 @@ test.describe("Draft75", function() { with(this) {

  itShouldBehaveLike("draft-75 protocol")
 }})
+
@@ -184,3 +184,4 @@ test.describe("Draft76", function() { with(this) {
    }})
  }})
 }})
+
@@ -199,42 +199,54 @@ 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]))
@@ -243,49 +255,67 @@ 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]))
@@ -294,27 +324,35 @@ 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) {
+      it("returns an error for too-large frames", function(resume) { with(this) {
        driver().parse([0x81, 0x7f, 0x00, 0x00, 0x00, 0x00, 0x40, 0x00, 0x00, 0x00])
-        assertEqual( "WebSocket frame length too large", error.message )
-        assertEqual( [1009, "WebSocket frame length too large"], close )
-        assertEqual( "closed", driver().getState() )
+        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 = []
@@ -323,12 +361,18 @@ 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)
      }})
    }})

@@ -379,18 +423,22 @@ 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)
      }})
    }})

@@ -427,15 +475,21 @@ 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)
    }})
  }})

@@ -479,8 +533,9 @@ 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) {
@@ -494,10 +549,11 @@ 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) {
@@ -539,3 +595,4 @@ test.describe("Hybi", function() { with(this) {
    }})
  }})
 }})
+
Author	SHA1	Message	Date
James Coglan	e6dea40c8c	[WIP] Begin implementing permessage-deflate.	2014-05-11 22:32:15 +01:00
James Coglan	7953bc39ff	Implement masking as a transform stream.	2014-05-11 22:07:34 +01:00
James Coglan	bf39b7b44b	Optimise some special cases in StreamReader; return a new Buffer from read() immediately if there's enough buffered data, and if the read can be fulfilled entirely from the first queued buffer than take a slice rather than copying into a new buffer.	2014-05-11 13:33:51 +01:00
James Coglan	85ba40ac32	Convert StreamReader so that it emits streams rather than waiting for an entire payload.	2014-05-09 15:35:40 +01:00
James Coglan	e6edb23681	Rather than a sync read() method in a loop, use an async read() method with a callback.	2014-05-08 16:35:21 +01:00