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.
Convert StreamReader so that it emits streams rather than waiting for an entire payload.
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
24 changed files with 447 additions and 227 deletions
@@ -8,3 +8,4 @@ node_js:

 before_install:
  - '[ "${TRAVIS_NODE_VERSION}" = "0.6" ] && npm conf set strict-ssl false || true'
+
@@ -1,12 +1,3 @@
-### 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
@@ -48,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
@@ -304,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.
+
@@ -18,3 +18,4 @@ var server = net.createServer(function(connection) {
 });

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

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

 module.exports = Base;
+
@@ -17,7 +17,7 @@ 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');
 };

@@ -109,3 +109,4 @@ for (var key in instance)
  Client.prototype[key] = instance[key];

 module.exports = Client;
+
@@ -115,3 +115,4 @@ for (var key in instance)
  Draft75.prototype[key] = instance[key];

 module.exports = Draft75;
+
@@ -106,3 +106,4 @@ for (var key in instance)
  Draft76.prototype[key] = instance[key];

 module.exports = Draft76;
+
@@ -27,3 +27,4 @@ Headers.prototype._strip = function(string) {
 };

 module.exports = Headers;
+
@@ -78,3 +78,4 @@ HttpParser.prototype.parse = function(data) {
 };

 module.exports = HttpParser;
+
@@ -1,47 +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 protos    = this._request.headers['sec-websocket-protocol'],
-      supported = this._protocols;
-
-  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.version) {
+    var version = this._request.headers['sec-websocket-version'];
+    this.version = 'hybi-' + version;
  }

-  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 || 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);
@@ -95,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) {
@@ -160,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);
@@ -209,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;
@@ -226,15 +174,26 @@ var instance = {
    var secKey = this._request.headers['sec-websocket-key'];
    if (!secKey) return '';

-    var headers = [
-      'HTTP/1.1 101 Switching Protocols',
-      'Upgrade: websocket',
-      'Connection: Upgrade',
-      'Sec-WebSocket-Accept: ' + Hybi.generateAccept(secKey)
-    ];
+    var accept    = Hybi.generateAccept(secKey),
+        protos    = this._request.headers['sec-websocket-protocol'],
+        supported = this._protocols,
+        proto,

-    if (this.protocol)
-      headers.push('Sec-WebSocket-Protocol: ' + this.protocol);
+        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);
+      }
+    }

    return new Buffer(headers.concat(this.__headers.toString(), '').join('\r\n'), 'utf8');
  },
@@ -242,7 +201,7 @@ var instance = {
  _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));
  },

@@ -251,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;
    });
@@ -262,8 +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._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);
@@ -274,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');
@@ -286,10 +250,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);
    }
  },

@@ -301,7 +265,7 @@ var instance = {

    if (!this._checkFrameLength()) return;

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

  _checkFrameLength: function() {
@@ -313,18 +277,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)
@@ -334,7 +313,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');
@@ -346,7 +325,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';
@@ -378,6 +357,8 @@ var instance = {
      delete callbacks[message];
      if (callback) callback()
    }
+
+    this._reader.read(1, this._parseOpcode);
  },

  _buffer: function(fragment) {
@@ -385,17 +366,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  = [];
@@ -422,3 +392,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];
@@ -40,9 +40,6 @@ var instance = {
      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 +101,4 @@ Server.http = function(request, options) {
 };

 module.exports = Server;
+
@@ -141,3 +141,4 @@ Messages.prototype.destroy = function() {};

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

-, "version"         : "0.3.6"
+, "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(
@@ -139,7 +131,10 @@ 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 )
@@ -157,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) {
@@ -176,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 )
@@ -190,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 )
@@ -204,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 )
@@ -220,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 )
@@ -239,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 )
@@ -257,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	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