When client handshake fails, close with the right code and reason.

Conflicts:
	lib/faye/websocket.js
	lib/faye/websocket/protocol8_parser.js

README.markdown

@@ -49,8 +49,8 @@ server.listen(8000);
 
 The client supports both the plain-text `ws` protocol and the encrypted `wss`
 protocol, and has exactly the same interface as a socket you would use in a web
-browser. On the wire it identifies itself as hybi-08, though it's compatible
-with servers speaking later versions of the protocol, at least up to version 17.
+browser. On the wire it identifies itself as hybi-13, though it's compatible
+with servers speaking later versions of the protocol.
 
 ```js
 var WebSocket = require('faye-websocket'),
@@ -72,6 +72,29 @@ ws.onclose = function(event) {
 ```
 
 
+## Subprotocol negotiation
+
+The WebSocket protocol allows peers to select and identify the application
+protocol to use over the connection. On the client side, you can set which
+protocols the client accepts by passing a list of protocol names when you
+construct the socket:
+
+```js
+var ws = new WebSocket.Client('ws://www.example.com/', ['irc', 'amqp']);
+```
+
+On the server side, you can likewise pass in the list of protocols the server
+supports after the other constructor arguments:
+
+```js
+var ws = new WebSocket(request, socket, head, ['irc', 'amqp']);
+```
+
+If the client and server agree on a protocol, both the client- and server-side
+socket objects expose the selected protocol through the `ws.protocol` property.
+If they cannot agree on a protocol to use, the client closes the connection.
+
+
 ## WebSocket API
 
 The WebSocket API consists of several event handlers and a method for sending
@@ -92,6 +115,8 @@ messages.
   sends a text or binary message over the connection to the other peer.
 * <b><tt>close(code, reason)</tt></b> closes the connection, sending the given
   status code and reason text, both of which are optional.
+* <b><tt>protocol</tt></b> is a string or `null` identifying the subprotocol the
+  socket is using.
 
 
 ## License

examples/autobahn_client.js

@@ -3,8 +3,7 @@ var WebSocket = require('../lib/faye/websocket');
 var host  = 'ws://localhost:9001',
     agent = 'Faye (Node ' + process.version + ')',
     cases = 0,
-    skip  = [247,248,249,250,251,252,253,254,255,
-             256,257,258,259,260,261,262,263,264];
+    skip  = [];
 
 var socket = new WebSocket.Client(host + '/getCaseCount');
 

examples/index.html

@@ -12,14 +12,19 @@
     <script type="text/javascript">
       var logger = document.getElementsByTagName('ul')[0],
           Socket = window.MozWebSocket || window.WebSocket,
-          socket = new Socket('ws://' + location.hostname + ':' + location.port + '/'),
+          protos = ['foo', 'bar', 'xmpp'],
+          socket = new Socket('ws://' + location.hostname + ':' + location.port + '/', protos),
           index  = 0;
       
       socket.onopen = function() {
-        logger.innerHTML += '<li>OPEN</li>';
+        logger.innerHTML += '<li>OPEN: ' + socket.protocol + '</li>';
         socket.send('Hello, world');
       };
       
+      socket.onerror = function(event) {
+        logger.innerHTML += '<li>ERROR: ' + error.message + '</li>';
+      };
+      
       socket.addEventListener('message', function(event) {
         logger.innerHTML += '<li>MESSAGE: ' + event.data + '</li>';
         setTimeout(function() { socket.send(++index + ' ' + event.data) }, 2000);

examples/server.js

@@ -25,8 +25,8 @@ var server = secure
            : http.createServer(staticHandler);
 
 server.addListener('upgrade', function(request, socket, head) {
-  var ws = new WebSocket(request, socket, head);
-  console.log('open', ws.url, ws.version);
+  var ws = new WebSocket(request, socket, head, ['irc', 'xmpp']);
+  console.log('open', ws.url, ws.version, ws.protocol);
   
   ws.onmessage = function(event) {
     ws.send(event.data);

lib/faye/websocket.js

@@ -30,7 +30,7 @@ var isSecureConnection = function(request) {
   }
 };
 
-var WebSocket = function(request, socket, head) {
+var WebSocket = function(request, socket, head, supportedProtos) {
   this.request = request;
   this._stream = request.socket;
   
@@ -40,9 +40,12 @@ var WebSocket = function(request, socket, head) {
   this.bufferedAmount = 0;
   
   var Parser = getParser(request);
-  this._parser = new Parser(this, this._stream);
-  this._parser.handshakeResponse(head);
+  this._parser = new Parser(this, {protocols: supportedProtos});
   
+  var handshake = this._parser.handshakeResponse(head);
+  try { this._stream.write(handshake, 'binary') } catch (e) {}
+  
+  this.protocol = this._parser.protocol;
   this.readyState = API.OPEN;
   this.version = this._parser.getVersion();
   
@@ -53,7 +56,9 @@ var WebSocket = function(request, socket, head) {
   var self = this;
   
   this._stream.addListener('data', function(data) {
-    self._parser.parse(data);
+    var response = self._parser.parse(data);
+    if (!response) return;
+    try { self._stream.write(response, 'binary') } catch (e) {}
   });
   ['close', 'end', 'error'].forEach(function(event) {
     self._stream.addListener(event, function() { self.close(1006, '', false) });

lib/faye/websocket/api.js

@@ -8,6 +8,7 @@ var API = {
   onmessage:  null,
   onerror:    null,
   onclose:    null,
+  protocol:   null,
   
   receive: function(data) {
     if (this.readyState !== API.OPEN) return false;
@@ -19,7 +20,13 @@ var API = {
   
   send: function(data, type, errorType) {
     if (this.readyState === API.CLOSED) return false;
-    return this._parser.frame(data, type, errorType);
+    var frame = this._parser.frame(data, type, errorType);
+    try {
+      this._stream.write(frame, 'binary');
+      return true;
+    } catch (e) {
+      return false;
+    }
   },
   
   close: function(code, reason, ack) {

lib/faye/websocket/client.js

@@ -4,22 +4,22 @@ var API = require('./api'),
 
 var Protocol8Parser = require('./protocol8_parser');
 
-var Client = function(url) {
-  this.url = url;
-  this.uri = require('url').parse(url);
+var Client = function(url, protocols) {
+  this.url  = url;
+  this._uri = require('url').parse(url);
   
   this.readyState = API.CONNECTING;
   this.bufferedAmount = 0;
   
-  var secure     = (this.uri.protocol === 'wss:'),
+  var secure     = (this._uri.protocol === 'wss:'),
       self       = this,
       onConnect  = function() { self._onConnect() },
       
       connection = secure
-                 ? tls.connect(this.uri.port || 443, this.uri.hostname, onConnect)
-                 : net.createConnection(this.uri.port || 80, this.uri.hostname);
+                 ? tls.connect(this._uri.port || 443, this._uri.hostname, onConnect)
+                 : net.createConnection(this._uri.port || 80, this._uri.hostname);
   
-  this._parser = new Protocol8Parser(this, connection, {masking: true});
+  this._parser = new Protocol8Parser(this, {masking: true, protocols: protocols});
   this._stream = connection;
   
   if (!secure) connection.addListener('connect', onConnect);
@@ -34,9 +34,11 @@ var Client = function(url) {
 };
   
 Client.prototype._onConnect = function() {
-  this._handshake = this._parser.createHandshake(this.uri);
+  this._handshake = this._parser.createHandshake(this._uri, this._stream);
   this._message = [];
-  this._handshake.requestData();
+  try {
+    this._stream.write(this._handshake.requestData(), 'binary');
+  } catch (e) {}
 };
 
 Client.prototype._onData = function(data) {
@@ -49,6 +51,7 @@ Client.prototype._onData = function(data) {
       if (!this._handshake.isComplete()) return;
       
       if (this._handshake.isValid()) {
+        this.protocol = this._handshake.protocol;
         this.readyState = API.OPEN;
         var event = new API.Event('open');
         event.initEvent('open', false, false);
@@ -58,7 +61,7 @@ Client.prototype._onData = function(data) {
         
       } else {
         this.readyState = API.CLOSED;
-        var event = new API.Event('close');
+        var event = new API.Event('close', {code: 1006, reason: ''});
         event.initEvent('close', false, false);
         this.dispatchEvent(event);
       }

lib/faye/websocket/draft75_parser.js

@@ -1,6 +1,5 @@
-var Draft75Parser = function(webSocket, stream) {
+var Draft75Parser = function(webSocket) {
   this._socket    = webSocket;
-  this._stream    = stream;
   this._buffer    = [];
   this._buffering = false;
 };
@@ -14,49 +13,43 @@ var instance = {
   },
   
   handshakeResponse: function() {
-    var stream = this._stream;
-    try {
-      stream.write( 'HTTP/1.1 101 Web Socket Protocol Handshake\r\n' +
-                    'Upgrade: WebSocket\r\n' +
-                    'Connection: Upgrade\r\n' +
-                    'WebSocket-Origin: ' + this._socket.request.headers.origin + '\r\n' +
-                    'WebSocket-Location: ' + this._socket.url + '\r\n\r\n');
-    } catch (e) {}
+    return new Buffer('HTTP/1.1 101 Web Socket Protocol Handshake\r\n' +
+                      'Upgrade: WebSocket\r\n' +
+                      'Connection: Upgrade\r\n' +
+                      'WebSocket-Origin: ' + this._socket.request.headers.origin + '\r\n' +
+                      'WebSocket-Location: ' + this._socket.url + '\r\n\r\n',
+                      'utf8');
   },
   
   parse: function(data) {
-    for (var i = 0, n = data.length; i < n; i++)
-      this._handleChar(data[i]);
+    for (var i = 0, n = data.length; i < n; i++) {
+      switch (data[i]) {
+        case 0x00:
+          this._buffering = true;
+          break;
+        
+        case 0xFF:
+          this._buffer = new Buffer(this._buffer);
+          this._socket.receive(this._buffer.toString('utf8', 0, this._buffer.length));
+          this._buffer = [];
+          this._buffering = false;
+          break;
+        
+        default:
+          if (this._buffering) this._buffer.push(data[i]);
+      }
+    }
   },
   
   frame: function(data) {
-    var stream = this._stream;
-    try {
-      stream.write(this.FRAME_START, 'binary');
-      stream.write(new Buffer(data), 'utf8');
-      stream.write(this.FRAME_END, 'binary');
-      return true;
-    } catch (e) {
-      return false;
-    }
-  },
-  
-  _handleChar: function(data) {
-    switch (data) {
-      case 0x00:
-        this._buffering = true;
-        break;
-      
-      case 0xFF:
-        this._buffer = new Buffer(this._buffer);
-        this._socket.receive(this._buffer.toString('utf8', 0, this._buffer.length));
-        this._buffer = [];
-        this._buffering = false;
-        break;
-      
-      default:
-        if (this._buffering) this._buffer.push(data);
-    }
+    var buffer = new Buffer(data, 'utf8'),
+        frame  = new Buffer(buffer.length + 2);
+    
+    this.FRAME_START.copy(frame, 0);
+    buffer.copy(frame, 1);
+    this.FRAME_END.copy(frame, buffer.length + 1);
+    
+    return frame;
   }
 };
 

lib/faye/websocket/draft76_parser.js

@@ -27,27 +27,31 @@ Draft76Parser.prototype.getVersion = function() {
 };
   
 Draft76Parser.prototype.handshakeResponse = function(head) {
-  var request = this._socket.request,
-      stream  = this._stream;
+  var request = this._socket.request, tmp;
   
-  try {
-    stream.write( 'HTTP/1.1 101 Web Socket Protocol Handshake\r\n' +
-                  'Upgrade: WebSocket\r\n' +
-                  'Connection: Upgrade\r\n' +
-                  'Sec-WebSocket-Origin: ' + request.headers.origin + '\r\n' +
-                  'Sec-WebSocket-Location: ' + this._socket.url + '\r\n\r\n',
-                  'binary');
-  } catch (e) {}
+  var response = new Buffer('HTTP/1.1 101 Web Socket Protocol Handshake\r\n' +
+                            'Upgrade: WebSocket\r\n' +
+                            'Connection: Upgrade\r\n' +
+                            'Sec-WebSocket-Origin: ' + request.headers.origin + '\r\n' +
+                            'Sec-WebSocket-Location: ' + this._socket.url + '\r\n\r\n',
+                            'binary');
   
-  this.handshakeSignature(head);
+  var signature = this.handshakeSignature(head);
+  if (signature) {
+    tmp = new Buffer(response.length + signature.length);
+    response.copy(tmp, 0);
+    signature.copy(tmp, response.length);
+    response = tmp;
+  }
+  
+  return response;
 };
 
 Draft76Parser.prototype.handshakeSignature = function(head) {
-  if (head.length === 0) return;
+  if (head.length === 0) return null;
   this._handshakeComplete = true;
   
   var request = this._socket.request,
-      stream  = this._stream,
       
       key1    = request.headers['sec-websocket-key1'],
       value1  = numberFromKey(key1) / spacesInKey(key1),
@@ -61,16 +65,14 @@ Draft76Parser.prototype.handshakeSignature = function(head) {
   MD5.update(bigEndian(value2));
   MD5.update(head.toString('binary'));
   
-  try {
-    stream.write(MD5.digest('binary'), 'binary');
-  } catch (e) {};
+  return new Buffer(MD5.digest('binary'), 'binary');
 };
 
 Draft76Parser.prototype.parse = function(data) {
   if (this._handshakeComplete)
     return Draft75Parser.prototype.parse.call(this, data);
   
-  this.handshakeSignature(data);
+  return this.handshakeSignature(data);
 };
 
 module.exports = Draft76Parser;

lib/faye/websocket/protocol8_parser.js

@@ -1,88 +1,31 @@
-var crypto = require('crypto');
+var crypto    = require('crypto'),
+    Handshake = require('./protocol8_parser/handshake'),
+    Reader    = require('./protocol8_parser/stream_reader');
 
-var Handshake = function(uri, stream) {
-  this._uri    = uri;
-  this._stream = stream;
-  
-  var buffer = new Buffer(16), i = 16;
-  while (i--) buffer[i] = Math.floor(Math.random() * 254);
-  this._key = buffer.toString('base64');
-  
-  var SHA1 = crypto.createHash('sha1');
-  SHA1.update(this._key + Protocol8Parser.prototype.GUID);
-  this._accept = SHA1.digest('base64');
-  
-  var HTTPParser = process.binding('http_parser').HTTPParser,
-      parser     = new HTTPParser(HTTPParser.RESPONSE || 'response'),
-      current    = null,
-      self       = this;
-  
-  this._nodeVersion = HTTPParser.RESPONSE ? 6 : 4;
-  this._complete    = false;
-  this._headers     = {};
-  this._parser      = parser;
-  
-  parser.onHeaderField = function(b, start, length) {
-    current = b.toString('utf8', start, start + length);
-  };
-  parser.onHeaderValue = function(b, start, length) {
-    self._headers[current] = b.toString('utf8', start, start + length);
-  };
-  parser.onHeadersComplete = function(info) {
-    self._status = info.statusCode;
-    var headers = info.headers;
-    if (!headers) return;
-    for (var i = 0, n = headers.length; i < n; i += 2)
-      self._headers[headers[i]] = headers[i+1];
-  };
-  parser.onMessageComplete = function() {
-    self._complete = true;
-  };
-};
-    
-Handshake.prototype.requestData = function() {
-  var stream = this._stream, u = this._uri;
-  try {
-    stream.write( 'GET ' + u.pathname + (u.search || '') + ' HTTP/1.1\r\n' +
-                  'Host: ' + u.hostname + (u.port ? ':' + u.port : '') + '\r\n' +
-                  'Upgrade: websocket\r\n' +
-                  'Connection: Upgrade\r\n' +
-                  'Sec-WebSocket-Key: ' + this._key + '\r\n' +
-                  'Sec-WebSocket-Version: 8\r\n\r\n');
-  } catch (e) {}
-};
-
-Handshake.prototype.parse = function(data) {
-  var consumed = this._parser.execute(data, 0, data.length),
-      offset   = (this._nodeVersion < 6) ? 1 : 0;
-  
-  return (consumed === data.length) ? [] : data.slice(consumed + offset);
-};
-
-Handshake.prototype.isComplete = function() {
-  return this._complete;
-};
-
-Handshake.prototype.isValid = function() {
-  if (this._status !== 101) return false;
-  
-  var upgrade    = this._headers.Upgrade,
-      connection = this._headers.Connection;
-  
-  return upgrade && /^websocket$/i.test(upgrade) &&
-         connection && connection.split(/\s*,\s*/).indexOf('Upgrade') >= 0 &&
-         this._headers['Sec-WebSocket-Accept'] === this._accept;
-};
-
-var Protocol8Parser = function(webSocket, stream, options) {
+var Protocol8Parser = function(webSocket, options) {
   this._reset();
-  this._socket  = webSocket;
-  this._stream  = stream;
-  this._stage   = 0;
-  this._masking = options && options.masking;
+  this._socket    = webSocket;
+  this._reader    = new Reader();
+  this._stage     = 0;
+  this._masking   = options && options.masking;
+  this._protocols = options && options.protocols;
+  
+  if (typeof this._protocols === 'string')
+    this._protocols = this._protocols.split(/\s*,\s*/);
+};
+
+Protocol8Parser.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;
 };
 
 var instance = {
+  BYTE:   255,
   FIN:    128,
   MASK:   128,
   RSV1:   64,
@@ -91,8 +34,6 @@ var instance = {
   OPCODE: 15,
   LENGTH: 127,
   
-  GUID: '258EAFA5-E914-47DA-95CA-C5AB0DC85B11',
-  
   OPCODES: {
     continuation: 0,
     text:         1,
@@ -127,102 +68,77 @@ var instance = {
   
   handshakeResponse: function() {
     var secKey = this._socket.request.headers['sec-websocket-key'];
-    if (!secKey) return;
+    if (!secKey) return null;
     
     var SHA1 = crypto.createHash('sha1');
-    SHA1.update(secKey + this.GUID);
-    var accept = SHA1.digest('base64');
+    SHA1.update(secKey + Handshake.GUID);
     
-    var stream = this._stream;
-    try {
-      stream.write( 'HTTP/1.1 101 Switching Protocols\r\n' +
-                    'Upgrade: websocket\r\n' +
-                    'Connection: Upgrade\r\n' +
-                    'Sec-WebSocket-Accept: ' + accept + '\r\n\r\n');
-    } catch (e) {}
+    var accept    = SHA1.digest('base64'),
+        protos    = this._socket.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 && supported !== 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('','').join('\r\n'), 'utf8');
   },
   
-  createHandshake: function() {
-    return new Handshake(this._socket.uri, this._stream);
+  createHandshake: function(uri) {
+    return new Handshake(uri, this._protocols);
   },
   
   parse: function(data) {
-    for (var i = 0, n = data.length; i < n; i++) {
+    this._reader.put(data);
+    var buffer = true;
+    while (buffer) {
       switch (this._stage) {
-        case 0: this._parseOpcode(data[i]);         break;
-        case 1: this._parseLength(data[i]);         break;
-        case 2: this._parseExtendedLength(data[i]); break;
-        case 3: this._parseMask(data[i]);           break;
-        case 4: this._parsePayload(data[i]);        break;
-      }
-      if (this._stage === 4 && this._length === 0)
-        this._emitFrame();
-    }
-  },
-  
-  frame: function(data, type, code) {
-    if (this._closed) return;
-    
-    var opcode = this.OPCODES[type || (typeof data === 'string' ? 'text' : 'binary')],
-        buffer = new Buffer(data),
-        insert = code ? 2 : 0,
-        length = buffer.length + insert,
-        masked = this._masking ? this.MASK : 0,
-        stream = this._stream,
-        frame, factor, mask, i, n;
-    
-    data = new Buffer(length);
-    if (code) {
-      data[0] = Math.floor(code / 256);
-      data[1] = code & 255;
-    }
-    for (i = 0, n = buffer.length; i < n; i++)
-      data[insert + i] = buffer[i];
-    
-    if (length <= 125) {
-      frame = new Buffer(2);
-      frame[1] = masked | length;
-    } else if (length >= 126 && length <= 65535) {
-      frame = new Buffer(4);
-      frame[1] = masked | 126;
-      frame[2] = Math.floor(length / 256);
-      frame[3] = length & 255;
-    } else {
-      frame = new Buffer(10);
-      frame[1] = masked | 127;
-      for (var i = 0; i < 8; i++) {
-        factor = Math.pow(2, 8 * (8 - 1 - i));
-        frame[2+i] = Math.floor(length / factor) & 255;
+        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;
       }
     }
-    frame[0] = this.FIN | opcode;
-    
-    if (this._masking) {
-      mask = [1,2,3,4].map(function() { return Math.floor(Math.random() * 255) });
-      for (i = 0, n = data.length; i < n; i++)
-        data[i] = data[i] ^ mask[i % 4];
-    }
-    
-    try {
-      stream.write(frame, 'binary');
-      if (mask) stream.write(new Buffer(mask), 'binary');
-      if (data.length > 0) stream.write(data, 'utf8');
-      return true;
-    } catch (e) {
-      return false;
-    }
-  },
-  
-  close: function(code, reason, callback, context) {
-    if (this._closed) return;
-    if (callback) this._closingCallback = [callback, context];
-    this.frame(reason || '', 'close', code || this.ERRORS.normal_closure);
-    this._closed = true;
-  },
-  
-  buffer: function(fragment) {
-    for (var i = 0, n = fragment.length; i < n; i++)
-      this._buffer.push(fragment[i]);
   },
   
   _parseOpcode: function(data) {
@@ -267,27 +183,75 @@ var instance = {
     }
   },
   
-  _parseExtendedLength: function(data) {
-    this._lengthBuffer.push(data);
-    if (this._lengthBuffer.length < this._lengthSize) return;
-    this._length = this._getInteger(this._lengthBuffer);
+  _parseExtendedLength: function(buffer) {
+    this._length = this._getInteger(buffer);
     this._stage  = this._masked ? 3 : 4;
   },
   
-  _parseMask: function(data) {
-    this._mask.push(data);
-    if (this._mask.length < 4) return;
-    this._stage = 4;
+  frame: function(data, type, code) {
+    if (this._closed) return null;
+    
+    var isText = (typeof data === 'string'),
+        opcode = this.OPCODES[type || (isText ? 'text' : 'binary')],
+        buffer = isText ? new Buffer(data, 'utf8') : data,
+        insert = code ? 2 : 0,
+        length = buffer.length + insert,
+        header = (length <= 125) ? 2 : (length <= 65535 ? 4 : 10),
+        offset = header + (this._masking ? 4 : 0),
+        masked = this._masking ? this.MASK : 0,
+        frame  = new Buffer(length + offset),
+        BYTE   = this.BYTE,
+        mask, i;
+    
+    frame[0] = this.FIN | opcode;
+    
+    if (length <= 125) {
+      frame[1] = masked | length;
+    } else if (length <= 65535) {
+      frame[1] = masked | 126;
+      frame[2] = Math.floor(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[9] = length & BYTE;
+    }
+    
+    if (code) {
+      frame[offset]   = Math.floor(code / 256) & BYTE;
+      frame[offset+1] = code & BYTE;
+    }
+    buffer.copy(frame, offset + insert);
+    
+    if (this._masking) {
+      mask = new Buffer([1,2,3,4].map(function() { return Math.floor(Math.random() * 256) }));
+      mask.copy(frame, header);
+      Protocol8Parser.mask(frame, mask, offset);
+    }
+    
+    return frame;
   },
   
-  _parsePayload: function(data) {
-    this._payload.push(data);
-    if (this._payload.length < this._length) return;
-    this._emitFrame();
+  close: function(code, reason, callback, context) {
+    if (this._closed) return;
+    if (callback) this._closingCallback = [callback, context];
+    this._socket.send(reason || '', 'close', code || this.ERRORS.normal_closure);
+    this._closed = true;
+  },
+  
+  buffer: function(fragment) {
+    for (var i = 0, n = fragment.length; i < n; i++)
+      this._buffer.push(fragment[i]);
   },
   
   _emitFrame: function() {
-    var payload = this._unmask(this._payload, this._mask),
+    var payload = Protocol8Parser.mask(this._payload, this._mask),
         opcode  = this._opcode;
     
     if (opcode === this.OPCODES.continuation) {
@@ -339,7 +303,6 @@ var instance = {
       if (payload.length > 125) return this._socket.close(this.ERRORS.protocol_error, null, false);
       this._socket.send(payload, 'pong');
     }
-    this._stage = 0;
   },
   
   _reset: function() {
@@ -360,16 +323,6 @@ var instance = {
     for (var i = 0, n = bytes.length; i < n; i++)
       number += bytes[i] << (8 * (n - 1 - i));
     return number;
-  },
-  
-  _unmask: function(payload, mask) {
-    var unmasked = new Buffer(payload.length), b;
-    for (var i = 0, n = payload.length; i < n; i++) {
-      b = payload[i];
-      if (mask.length > 0) b = b ^ mask[i % 4];
-      unmasked[i] = b;
-    }
-    return unmasked;
   }
 };
 

lib/faye/websocket/protocol8_parser/handshake.js

@@ -0,0 +1,91 @@
+var crypto = require('crypto');
+
+var Handshake = function(uri, protocols) {
+  this._uri = uri;
+  this._protocols = protocols;
+  
+  var buffer = new Buffer(16), i = 16;
+  while (i--) buffer[i] = Math.floor(Math.random() * 256);
+  this._key = buffer.toString('base64');
+  
+  var SHA1 = crypto.createHash('sha1');
+  SHA1.update(this._key + Handshake.GUID);
+  this._accept = SHA1.digest('base64');
+  
+  var HTTPParser = process.binding('http_parser').HTTPParser,
+      parser     = new HTTPParser(HTTPParser.RESPONSE || 'response'),
+      current    = null,
+      self       = this;
+  
+  this._nodeVersion = HTTPParser.RESPONSE ? 6 : 4;
+  this._complete    = false;
+  this._headers     = {};
+  this._parser      = parser;
+  
+  parser.onHeaderField = function(b, start, length) {
+    current = b.toString('utf8', start, start + length);
+  };
+  parser.onHeaderValue = function(b, start, length) {
+    self._headers[current] = b.toString('utf8', start, start + length);
+  };
+  parser.onHeadersComplete = function(info) {
+    self._status = info.statusCode;
+    var headers = info.headers;
+    if (!headers) return;
+    for (var i = 0, n = headers.length; i < n; i += 2)
+      self._headers[headers[i]] = headers[i+1];
+  };
+  parser.onMessageComplete = function() {
+    self._complete = true;
+  };
+};
+
+Handshake.GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11';
+    
+Handshake.prototype.requestData = function() {
+  var u = this._uri;
+  
+  var headers = [
+    'GET ' + u.pathname + (u.search || '') + ' HTTP/1.1',
+    'Host: ' + u.hostname + (u.port ? ':' + u.port : ''),
+    'Upgrade: websocket',
+    'Connection: Upgrade',
+    'Sec-WebSocket-Key: ' + this._key,
+    'Sec-WebSocket-Version: 13'
+  ];
+  
+  if (this._protocols)
+    headers.push('Sec-WebSocket-Protocol: ' + this._protocols.join(', '));
+  
+  return new Buffer(headers.concat('','').join('\r\n'), 'utf8');
+};
+
+Handshake.prototype.parse = function(data) {
+  var consumed = this._parser.execute(data, 0, data.length),
+      offset   = (this._nodeVersion < 6) ? 1 : 0;
+  
+  return (consumed === data.length) ? [] : data.slice(consumed + offset);
+};
+
+Handshake.prototype.isComplete = function() {
+  return this._complete;
+};
+
+Handshake.prototype.isValid = function() {
+  if (this._status !== 101) return false;
+  
+  var upgrade    = this._headers.Upgrade,
+      connection = this._headers.Connection,
+      protocol   = this._headers['Sec-WebSocket-Protocol'];
+  
+  this.protocol = this._protocols && this._protocols.indexOf(protocol) >= 0
+                ? protocol
+                : null;
+  
+  return upgrade && /^websocket$/i.test(upgrade) &&
+         connection && connection.split(/\s*,\s*/).indexOf('Upgrade') >= 0 &&
+         ((!this._protocols && !protocol) || this.protocol) &&
+         this._headers['Sec-WebSocket-Accept'] === this._accept;
+};
+
+module.exports = Handshake;

lib/faye/websocket/protocol8_parser/stream_reader.js

@@ -0,0 +1,43 @@
+var StreamReader = function() {
+  this._queue  = [];
+  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);
+  this._queue.push(buffer);
+};
+
+StreamReader.prototype._readBuffer = function(length) {
+  var buffer = new Buffer(length),
+      queue  = this._queue,
+      remain = length,
+      n      = queue.length,
+      i      = 0,
+      chunk, offset, size;
+  
+  if (remain === 0) return buffer;
+  
+  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;
+    i += 1;
+  }
+  
+  if (remain > 0) return null;
+  
+  queue.splice(0, i-1);
+  this._cursor = (i === 1 ? this._cursor : 0) + size;
+  
+  return buffer;
+};
+
+module.exports = StreamReader;

package.json

@@ -1,5 +1,5 @@
 { "name"            : "faye-websocket"
-, "description"     : "Robust general-purpose WebSocket server and client"
+, "description"     : "Standards-compliant WebSocket server and client"
 , "homepage"        : "http://github.com/jcoglan/faye-websocket-node"
 , "author"          : "James Coglan <jcoglan@gmail.com> (http://jcoglan.com/)"
 , "keywords"        : ["websocket"]

spec/faye/websocket/client_spec.js

@@ -13,7 +13,7 @@ JS.ENV.WebSocketSteps = JS.Test.asyncSteps({
     setTimeout(callback, 100)
   },
   
-  open_socket: function(url, callback) {
+  open_socket: function(url, protocols, callback) {
     var done = false,
         self = this,
         
@@ -24,7 +24,7 @@ JS.ENV.WebSocketSteps = JS.Test.asyncSteps({
                    callback()
                  }
     
-    this._ws = new Client(url)
+    this._ws = new Client(url, protocols)
     
     this._ws.onopen  = function() { resume(true)  }
     this._ws.onclose = function() { resume(false) }
@@ -49,6 +49,11 @@ JS.ENV.WebSocketSteps = JS.Test.asyncSteps({
     callback()
   },
   
+  check_protocol: function(protocol, callback) {
+    this.assertEqual( protocol, this._ws.protocol )
+    callback()
+  },
+  
   listen_for_message: function(callback) {
     var self = this
     this._ws.addEventListener('message', function(message) { self._message = message.data })
@@ -76,30 +81,37 @@ JS.ENV.ClientSpec = JS.Test.describe("Client", function() { with(this) {
   include(WebSocketSteps)
   
   before(function() {
+    this.protocols       = ["foo", "echo"]
     this.plain_text_url  = "ws://localhost:8000/bayeux"
     this.secure_url      = "wss://localhost:8000/bayeux"
   })
   
   sharedBehavior("socket client", function() { with(this) {
     it("can open a connection", function() { with(this) {
-      open_socket(socket_url)
+      open_socket(socket_url, protocols)
       check_open()
+      check_protocol("echo")
     }})
     
     it("cannot open a connection to the wrong host", function() { with(this) {
-      open_socket(blocked_url)
+      open_socket(blocked_url, protocols)
+      check_closed()
+    }})
+    
+    it("cannot open a connection with unacceptable protocols", function() { with(this) {
+      open_socket(socket_url, ["foo"])
       check_closed()
     }})
     
     it("can close the connection", function() { with(this) {
-      open_socket(socket_url)
+      open_socket(socket_url, protocols)
       close_socket()
       check_closed()
     }})
     
     describe("in the OPEN state", function() { with(this) {
       before(function() { with(this) {
-        open_socket(socket_url)
+        open_socket(socket_url, protocols)
       }})
       
       it("can send and receive messages", function() { with(this) {
@@ -111,7 +123,7 @@ JS.ENV.ClientSpec = JS.Test.describe("Client", function() { with(this) {
     
     describe("in the CLOSED state", function() { with(this) {
       before(function() { with(this) {
-        open_socket(socket_url)
+        open_socket(socket_url, protocols)
         close_socket()
       }})
       

spec/faye/websocket/draft75parser_spec.js

@@ -3,8 +3,7 @@ var Draft75Parser = require('../../../lib/faye/websocket/draft75_parser')
 JS.ENV.Draft75ParserSpec = JS.Test.describe("Draft75Parser", function() { with(this) {
   before(function() { with(this) {
     this.webSocket = {dispatchEvent: function() {}}
-    this.socket = new FakeSocket
-    this.parser = new Draft75Parser(webSocket, socket)
+    this.parser = new Draft75Parser(webSocket)
   }})
   
   describe("parse", function() { with(this) {
@@ -27,13 +26,11 @@ JS.ENV.Draft75ParserSpec = JS.Test.describe("Draft75Parser", function() { with(t
   
   describe("frame", function() { with(this) {
     it("returns the given string formatted as a WebSocket frame", function() { with(this) {
-      parser.frame("Hello")
-      assertEqual( [0x00, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0xff], socket.read() )
+      assertBufferEqual( [0x00, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0xff], parser.frame("Hello") )
     }})
     
     it("encodes multibyte characters correctly", function() { with(this) {
-      parser.frame("Apple = ")
-      assertEqual( [0x00, 0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3, 0xbf, 0xff], socket.read() )
+      assertBufferEqual( [0x00, 0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3, 0xbf, 0xff], parser.frame("Apple = ") )
     }})
   }})
 }})

spec/faye/websocket/protocol8parser_spec.js

@@ -3,8 +3,7 @@ var Protocol8Parser = require('../../../lib/faye/websocket/protocol8_parser')
 JS.ENV.Protocol8ParserSpec = JS.Test.describe("Protocol8Parser", function() { with(this) {
   before(function() { with(this) {
     this.webSocket = {dispatchEvent: function() {}}
-    this.socket = new FakeSocket
-    this.parser = new Protocol8Parser(webSocket, socket)
+    this.parser = new Protocol8Parser(webSocket)
   }})
   
   define("parse", function() {
@@ -116,18 +115,15 @@ JS.ENV.Protocol8ParserSpec = JS.Test.describe("Protocol8Parser", function() { wi
   
   describe("frame", function() { with(this) {
     it("returns the given string formatted as a WebSocket frame", function() { with(this) {
-      parser.frame("Hello")
-      assertEqual( [0x81, 0x05, 0x48, 0x65, 0x6c, 0x6c, 0x6f], socket.read() )
+      assertBufferEqual( [0x81, 0x05, 0x48, 0x65, 0x6c, 0x6c, 0x6f], parser.frame("Hello") )
     }})
     
     it("encodes multibyte characters correctly", function() { with(this) {
-      parser.frame("Apple = ")
-      assertEqual( [0x81, 0x0b, 0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3, 0xbf], socket.read() )
+      assertBufferEqual( [0x81, 0x0b, 0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3, 0xbf], parser.frame("Apple = ") )
     }})
     
     it("encodes medium-length strings using extra length bytes", function() { with(this) {
-      parser.frame("HelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHello")
-      assertEqual( [129, 126, 0, 200, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111], socket.read() )
+      assertBufferEqual( [129, 126, 0, 200, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111, 72, 101, 108, 108, 111], parser.frame("HelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHelloHello") )
     }})
     
     it("encodes long strings using extra length bytes", function() { with(this) {
@@ -136,18 +132,15 @@ JS.ENV.Protocol8ParserSpec = JS.Test.describe("Protocol8Parser", function() { wi
         message += "Hello"
         output = output.concat([0x48, 0x65, 0x6c, 0x6c, 0x6f])
       }
-      parser.frame(message)
-      assertEqual( output, socket.read() )
+      assertBufferEqual( output, parser.frame(message) )
     }})
     
     it("encodes close frames with an error code", function() { with(this) {
-      parser.frame("Hello", "close", 1002)
-      assertEqual( [0x88, 0x07, 0x03, 0xea, 0x48, 0x65, 0x6c, 0x6c, 0x6f], socket.read() )
+      assertBufferEqual( [0x88, 0x07, 0x03, 0xea, 0x48, 0x65, 0x6c, 0x6c, 0x6f], parser.frame("Hello", "close", 1002) )
     }})
     
     it("encodes pong frames", function() { with(this) {
-      parser.frame("", "pong")
-      assertEqual( [0x8a, 0x00], socket.read() )
+      assertBufferEqual( [0x8a, 0x00], parser.frame("", "pong") )
     }})
   }})
 }})

spec/runner.js

@@ -6,23 +6,6 @@ var WebSocket = require('../lib/faye/websocket'),
     https     = require('https')
 
 
-JS.ENV.FakeSocket = function() {
-  this._fragments = []
-}
-FakeSocket.prototype.write = function(buffer, encoding) {
-  this._fragments.push([buffer, encoding])
-}
-FakeSocket.prototype.read = function() {
-  var output = []
-  this._fragments.forEach(function(buffer, i) {
-    for (var j = 0, n = buffer[0].length; j < n; j++)
-    output.push(buffer[0][j])
-  })
-  return output
-}
-FakeSocket.prototype.addListener = function() {}
-
-
 JS.ENV.EchoServer = function() {}
 EchoServer.prototype.listen = function(port, ssl) {
   var server = ssl
@@ -33,7 +16,7 @@ EchoServer.prototype.listen = function(port, ssl) {
              : http.createServer()
   
   server.addListener('upgrade', function(request, socket, head) {
-    var ws = new WebSocket(request, socket, head)
+    var ws = new WebSocket(request, socket, head, ["echo"])
     ws.onmessage = function(event) {
       ws.send(event.data)
     }
@@ -55,6 +38,13 @@ JS.Packages(function() { with(this) {
 
 
 JS.require('JS.Test', function() {
+  JS.Test.Unit.Assertions.define("assertBufferEqual", function(array, buffer) {
+    this.assertEqual(array.length, buffer.length);
+    var ary = [], n = buffer.length;
+    while (n--) ary[n] = buffer[n];
+    this.assertEqual(array, ary);
+  })
+  
   JS.require( 'ClientSpec',
               'Draft75ParserSpec',
               'Protocol8ParserSpec',