Provide usable and accurate (HAProxy-safe) onopen event on the server…

… side.

examples/server.js

@@ -10,6 +10,10 @@ var upgradeHandler = function(request, socket, head) {
   var ws = new WebSocket(request, socket, head, ['irc', 'xmpp']);
   console.log('open', ws.url, ws.version, ws.protocol);
 
+  ws.onopen = function() {
+    ws.send('The server says hi.');
+  };
+
   ws.onmessage = function(event) {
     ws.send(event.data);
   };

lib/faye/eventsource.js

@@ -26,13 +26,16 @@ var EventSource = function(request, response, options) {
   this.url = scheme + '//' + request.headers.host + request.url;
 
   this.lastEventId = request.headers['last-event-id'] || '';
-  this.readyState  = API.OPEN;
-
-  var event = new Event('open');
-  event.initEvent('open', false, false);
-  this.dispatchEvent(event);
 
   var self = this;
+
+  process.nextTick(function() {
+    self.readyState  = API.OPEN;
+    var event = new Event('open');
+    event.initEvent('open', false, false);
+    self.dispatchEvent(event);
+  });
+
   this._pingLoop = setInterval(function() {
     try { this._stream.write(':\r\n\r\n') } catch (e) {}
   }, this._ping * 1000);

lib/faye/websocket.js

@@ -50,19 +50,25 @@ var WebSocket = function(request, socket, head, supportedProtos) {
   try { this._stream.write(handshake, 'binary') } catch (e) {}
 
   this.protocol = this._parser.protocol || '';
-  this.readyState = API.OPEN;
   this.version = this._parser.getVersion();
 
-  var event = new Event('open');
-  event.initEvent('open', false, false);
-  this.dispatchEvent(event);
-
   var self = this;
 
+  var open = function() {
+    if (!self._parser.isOpen()) return;
+    self.readyState = API.OPEN;
+    var event = new Event('open');
+    event.initEvent('open', false, false);
+    self.dispatchEvent(event);
+  };
+
+  process.nextTick(open);
+
   this._stream.addListener('data', function(data) {
     var response = self._parser.parse(data);
     if (!response) return;
     try { self._stream.write(response, 'binary') } catch (e) {}
+    open();
   });
   ['close', 'end', 'error'].forEach(function(event) {
     self._stream.addListener(event, function() { self.close(1006, '', false) });

lib/faye/websocket/draft75_parser.js

@@ -17,6 +17,10 @@ var instance = {
                       'utf8');
   },
 
+  isOpen: function() {
+    return true;
+  },
+
   parse: function(buffer) {
     var data, message, value;
     for (var i = 0, n = buffer.length; i < n; i++) {

lib/faye/websocket/draft76_parser.js

@@ -47,9 +47,12 @@ Draft76Parser.prototype.handshakeResponse = function(head) {
   return response;
 };
 
+Draft76Parser.prototype.isOpen = function() {
+  return !!this._handshakeComplete;
+};
+
 Draft76Parser.prototype.handshakeSignature = function(head) {
   if (head.length === 0) return null;
-  this._handshakeComplete = true;
 
   var request = this._socket.request,
 
@@ -65,6 +68,7 @@ Draft76Parser.prototype.handshakeSignature = function(head) {
   MD5.update(bigEndian(value2));
   MD5.update(head.toString('binary'));
 
+  this._handshakeComplete = true;
   return new Buffer(MD5.digest('binary'), 'binary');
 };
 

lib/faye/websocket/hybi_parser.js

@@ -97,6 +97,10 @@ var instance = {
     return new Buffer(headers.concat('','').join('\r\n'), 'utf8');
   },
 
+  isOpen: function() {
+    return true;
+  },
+
   createHandshake: function(uri) {
     return new Handshake(uri, this._protocols);
   },