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
7 changed files with 396 additions and 181 deletions
@@ -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');
 };

@@ -1,14 +1,15 @@
-var crypto   = require('crypto'),
-    util     = require('util'),
-    Base     = require('./base'),
-    Reader   = require('./hybi/stream_reader');
+var crypto = require('crypto'),
+    util   = require('util'),
+    Base   = require('./base'),
+    Concat = require('./hybi/concat'),
+    Mask   = require('./hybi/mask'),
+    Reader = require('./hybi/stream_reader');

 var Hybi = function(request, url, options) {
  Base.apply(this, arguments);
  this._reset();

-  this._reader    = new Reader();
-  this._stage     = 0;
+  this._reader    = new Reader({context: this});
  this._masking   = this._options.masking;
  this._protocols = this._options.protocols || [];

@@ -22,19 +23,11 @@ var Hybi = function(request, url, options) {
    var version = this._request.headers['sec-websocket-version'];
    this.version = 'hybi-' + version;
  }
+
+  this._reader.read(1, this._parseOpcode);
 };
 util.inherits(Hybi, Base);

-Hybi.mask = function(payload, mask, offset) {
-  if (mask.length === 0) return payload;
-  offset = offset || 0;
-
-  for (var i = 0, n = payload.length - offset; i < n; i++) {
-    payload[offset + i] = payload[offset + i] ^ mask[i % 4];
-  }
-  return payload;
-};
-
 Hybi.generateAccept = function(key) {
  var sha1 = crypto.createHash('sha1');
  sha1.update(key + Hybi.GUID);
@@ -88,46 +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._payload = buffer;
-            this._emitFrame();
-            this._stage = 0;
-          }
-          break;
-
-        default:
-          buffer = null;
-      }
-    }
+    this._reader.write(data);
  },

  frame: function(data, type, code) {
@@ -154,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);
@@ -247,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));
  },

@@ -256,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;
    });
@@ -267,10 +223,9 @@ var instance = {
          ', reserved2 = ' + (rsvs[1] ? 1 : 0) +
          ', reserved3 = ' + (rsvs[2] ? 1 : 0));

-    this._final   = (data & this.FIN) === this.FIN;
-    this._opcode  = (data & this.OPCODE);
-    this._mask    = [];
-    this._payload = [];
+    this._final  = (data & this.FIN) === this.FIN;
+    this._opcode = (data & this.OPCODE);
+    this._mask   = null;

    if (this.OPCODE_CODES.indexOf(this._opcode) < 0)
      return this._fail('protocol_error', 'Unrecognized frame opcode: ' + this._opcode);
@@ -281,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');
@@ -293,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);
    }
  },

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

    if (!this._checkFrameLength()) return;

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

  _checkFrameLength: function() {
@@ -320,15 +277,33 @@ var instance = {
    }
  },

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

    if (opcode === this.OPCODES.continuation) {
      if (!this._mode) return this._fail('protocol_error', 'Received unexpected continuation frame');
      this._buffer(payload);
      if (this._final) {
-        var message = this._concatBuffer();
+        var message = Concat.concatBuffers(this.__buffer, this.__blength);
        if (this._mode === 'text') message = this._encode(message);
        this._reset();
        if (message === null)
@@ -382,6 +357,8 @@ var instance = {
      delete callbacks[message];
      if (callback) callback()
    }
+
+    this._reader.read(1, this._parseOpcode);
  },

  _buffer: function(fragment) {
@@ -389,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  = [];
@@ -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];
@@ -131,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 )
@@ -149,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) {
@@ -168,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 )
@@ -182,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 )
@@ -196,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 )
@@ -212,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 )
@@ -231,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 )
@@ -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) {
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