Bump version to 0.3.4.

StreamReader: Don't hang on to fully read buffers

.travis.yml

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

CHANGELOG.md

@@ -1,3 +1,17 @@
+### 0.3.4 / 2014-05-08
+
+* Don't hold memory-leaking references to I/O buffers after they have been parsed
+
+### 0.3.3 / 2014-04-24
+
+* Correct the draft-76 status line reason phrase
+
+### 0.3.2 / 2013-12-29
+
+* Expand `maxLength` to cover sequences of continuation frames and `draft-{75,76}`
+* Decrease default maximum frame buffer size to 64MB
+* Stop parsing when the protocol enters a failure mode, to save CPU cycles
+
 ### 0.3.1 / 2013-12-03
 
 * Add a `maxLength` option to limit allowed frame size

README.md

@@ -174,7 +174,7 @@ The `options` argument is optional, and is an object. It may contain the
 following fields:
 
 * `maxLength` - the maximum allowed size of incoming message frames, in bytes.
-  The default value is `2^30 - 1`, or 1 byte short of 1 GiB.
+  The default value is `2^26 - 1`, or 1 byte short of 64 MiB.
 * `protocols` - an array of strings representing acceptable subprotocols for
   use over the socket. The driver will negotiate one of these to use via the
   `Sec-WebSocket-Protocol` header if supported by the other peer.
@@ -285,7 +285,7 @@ after `emit('open')` has fired.
 
 (The MIT License)
 
-Copyright (c) 2010-2013 James Coglan
+Copyright (c) 2010-2014 James Coglan
 
 Permission is hereby granted, free of charge, to any person obtaining a copy of
 this software and associated documentation files (the 'Software'), to deal in

lib/websocket/driver/base.js

@@ -8,6 +8,7 @@ var Base = function(request, url, options) {
 
   this._request   = request;
   this._options   = options || {};
+  this._maxLength = this._options.maxLength || this.MAX_LENGTH;
   this.__headers  = new Headers();
   this.__queue    = [];
   this.readyState = 0;
@@ -20,6 +21,10 @@ var Base = function(request, url, options) {
 util.inherits(Base, Emitter);
 
 var instance = {
+  // This is 64MB, small enough for an average VPS to handle without
+  // crashing from process out of memory
+  MAX_LENGTH: 0x3ffffff,
+
   STATES: ['connecting', 'open', 'closing', 'closed'],
 
   _bindEventListeners: function() {

lib/websocket/driver/draft75.js

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

lib/websocket/driver/draft76.js

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

lib/websocket/driver/hybi.js

@@ -11,7 +11,6 @@ var Hybi = function(request, url, options) {
   this._stage     = 0;
   this._masking   = this._options.masking;
   this._protocols = this._options.protocols || [];
-  this._maxLength = this._options.maxLength || this.MAX_LENGTH;
 
   if (typeof this._protocols === 'string')
     this._protocols = this._protocols.split(/\s*,\s*/);
@@ -67,9 +66,6 @@ var instance = {
   FRAGMENTED_OPCODES: [0, 1, 2],
   OPENING_OPCODES:    [1, 2],
 
-  // This is the maximum length of a Node buffer:
-  // https://github.com/joyent/node/blob/master/src/smalloc.h#L40
-  MAX_LENGTH: 0x3fffffff,
   TWO_POWERS: [0, 1, 2, 3, 4, 5, 6, 7].map(function(n) { return Math.pow(2, 8 * n) }),
 
   ERRORS: {
@@ -127,6 +123,9 @@ var instance = {
             this._stage = 0;
           }
           break;
+
+        default:
+          buffer = null;
       }
     }
   },
@@ -248,6 +247,7 @@ var instance = {
   _shutdown: function(code, reason) {
     this.frame(reason, 'close', code);
     this.readyState = 3;
+    this._stage = 5;
     this.emit('close', new Base.CloseEvent(code, reason));
   },
 
@@ -292,11 +292,11 @@ var instance = {
     this._length = (data & this.LENGTH);
 
     if (this._length >= 0 && this._length <= 125) {
+      if (!this._checkFrameLength()) return;
       this._stage = this._masked ? 3 : 4;
     } else {
-      this._lengthBuffer = [];
-      this._lengthSize   = (this._length === 126 ? 2 : 8);
-      this._stage        = 2;
+      this._lengthSize = (this._length === 126 ? 2 : 8);
+      this._stage      = 2;
     }
   },
 
@@ -306,12 +306,20 @@ var instance = {
     if (this.FRAGMENTED_OPCODES.indexOf(this._opcode) < 0 && this._length > 125)
       return this._fail('protocol_error', 'Received control frame having too long payload: ' + this._length);
 
-    if (this._length > this._maxLength)
-      return this._fail('too_large', 'WebSocket frame length too large');
+    if (!this._checkFrameLength()) return;
 
     this._stage  = this._masked ? 3 : 4;
   },
 
+  _checkFrameLength: function() {
+    if (this.__blength + this._length > this._maxLength) {
+      this._fail('too_large', 'WebSocket frame length too large');
+      return false;
+    } else {
+      return true;
+    }
+  },
+
   _emitFrame: function() {
     var payload = Hybi.mask(this._payload, this._mask),
         opcode  = this._opcode;
@@ -320,7 +328,7 @@ var instance = {
       if (!this._mode) return this._fail('protocol_error', 'Received unexpected continuation frame');
       this._buffer(payload);
       if (this._final) {
-        var message = new Buffer(this.__buffer);
+        var message = this._concatBuffer();
         if (this._mode === 'text') message = this._encode(message);
         this._reset();
         if (message === null)
@@ -377,13 +385,25 @@ var instance = {
   },
 
   _buffer: function(fragment) {
-    for (var i = 0, n = fragment.length; i < n; i++)
-      this.__buffer.push(fragment[i]);
+    this.__buffer.push(fragment);
+    this.__blength += fragment.length;
+  },
+
+  _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 = [];
+    this._mode     = null;
+    this.__buffer  = [];
+    this.__blength = 0;
   },
 
   _encode: function(buffer) {

lib/websocket/driver/hybi/stream_reader.js

@@ -4,10 +4,6 @@ var StreamReader = function() {
   this._cursor    = 0;
 };
 
-StreamReader.prototype.read = function(bytes) {
-  return this._readBuffer(bytes);
-};
-
 StreamReader.prototype.put = function(buffer) {
   if (!buffer || buffer.length === 0) return;
   if (!buffer.copy) buffer = new Buffer(buffer);
@@ -15,7 +11,7 @@ StreamReader.prototype.put = function(buffer) {
   this._queueSize += buffer.length;
 };
 
-StreamReader.prototype._readBuffer = function(length) {
+StreamReader.prototype.read = function(length) {
   if (length > this._queueSize) return null;
 
   var buffer = new Buffer(length),
@@ -23,22 +19,22 @@ StreamReader.prototype._readBuffer = function(length) {
       remain = length,
       n      = queue.length,
       i      = 0,
-      chunk, offset, size;
-
-  if (remain === 0) return buffer;
+      chunk, size;
 
   while (remain > 0 && i < n) {
     chunk = queue[i];
-    offset = (i === 0) ? this._cursor : 0;
-    size = Math.min(remain, chunk.length - offset);
-    chunk.copy(buffer, length - remain, offset, offset + size);
-    remain -= size;
+    size  = Math.min(remain, chunk.length - this._cursor);
+
+    chunk.copy(buffer, length - remain, this._cursor, this._cursor + size);
+
+    remain          -= size;
     this._queueSize -= size;
+    this._cursor     = (this._cursor + size) % chunk.length;
+
     i += 1;
   }
 
-  queue.splice(0, i-1);
-  this._cursor = (i === 1 ? this._cursor : 0) + size;
+  queue.splice(0, this._cursor === 0 ? i : i - 1);
 
   return buffer;
 };

package.json

@@ -5,7 +5,7 @@
 , "keywords"        : ["websocket"]
 , "license"         : "MIT"
 
-, "version"         : "0.3.1"
+, "version"         : "0.3.4"
 , "engines"         : {"node": ">=0.4.0"}
 , "main"            : "./lib/websocket/driver"
 , "devDependencies" : {"jstest": ""}

spec/websocket/driver/draft76_spec.js

@@ -53,7 +53,7 @@ test.describe("Draft76", function() { with(this) {
     describe("start", function() { with(this) {
       it("writes the handshake response to the socket", function() { with(this) {
         expect(driver().io, "emit").given("data", buffer(
-            "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" +
+            "HTTP/1.1 101 WebSocket Protocol Handshake\r\n" +
             "Upgrade: WebSocket\r\n" +
             "Connection: Upgrade\r\n" +
             "Sec-WebSocket-Origin: http://www.example.com\r\n" +
@@ -95,7 +95,7 @@ test.describe("Draft76", function() { with(this) {
 
       it("queues the frames until the handshake has been sent", function() { with(this) {
         expect(driver().io, "emit").given("data", buffer(
-            "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" +
+            "HTTP/1.1 101 WebSocket Protocol Handshake\r\n" +
             "Upgrade: WebSocket\r\n" +
             "Connection: Upgrade\r\n" +
             "Sec-WebSocket-Origin: http://www.example.com\r\n" +
@@ -116,7 +116,7 @@ test.describe("Draft76", function() { with(this) {
 
       it("writes the handshake response with no body", function() { with(this) {
         expect(driver().io, "emit").given("data", buffer(
-            "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" +
+            "HTTP/1.1 101 WebSocket Protocol Handshake\r\n" +
             "Upgrade: WebSocket\r\n" +
             "Connection: Upgrade\r\n" +
             "Sec-WebSocket-Origin: http://www.example.com\r\n" +