Re-apply "Add a non-dart:io WebSocket implementation to http_parser."

pkg/http_parser/lib/src/web_socket.dart

-// Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
+// Copyright (c) 2014, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
-part of dart.io;
+library http_parser.web_socket;
+
+import 'dart:async';
+import 'dart:convert';
+import 'dart:math';
+import 'dart:typed_data';
+
+import 'package:crypto/crypto.dart';
+
+import 'bytes_builder.dart';
+
+/// An implementation of the WebSocket protocol that's not specific to "dart:io"
+/// or to any particular HTTP API.
+///
+/// Because this is HTTP-API-agnostic, it doesn't handle the initial [WebSocket
+/// handshake][]. This needs to be handled manually by the user of the code.
+/// Once that's been done, [new CompatibleWebSocket] can be called with the
+/// underlying socket and it will handle the remainder of the protocol.
+///
+/// [WebSocket handshake]: https://tools.ietf.org/html/rfc6455#section-4
+abstract class CompatibleWebSocket implements Stream, StreamSink {
+  /// The interval for sending ping signals.
+  ///
+  /// If a ping message is not answered by a pong message from the peer, the
+  /// `WebSocket` is assumed disconnected and the connection is closed with a
+  /// [WebSocketStatus.GOING_AWAY] close code. When a ping signal is sent, the
+  /// pong message must be received within [pingInterval].
+  ///
+  /// There are never two outstanding pings at any given time, and the next ping
+  /// timer starts when the pong is received.
+  ///
+  /// By default, the [pingInterval] is `null`, indicating that ping messages
+  /// are disabled.
+  Duration pingInterval;
+
+  /// The [close code][] set when the WebSocket connection is closed.
+  ///
+  /// [close code]: https://tools.ietf.org/html/rfc6455#section-7.1.5
+  ///
+  /// Before the connection has been closed, this will be `null`.
+  int get closeCode;
+
+  /// The [close reason][] set when the WebSocket connection is closed.
+  ///
+  /// [close reason]: https://tools.ietf.org/html/rfc6455#section-7.1.6
+  ///
+  /// Before the connection has been closed, this will be `null`.
+  String get closeReason;
+
+  /// Signs a `Sec-WebSocket-Key` header sent by a WebSocket client as part of
+  /// the [initial handshake].
+  ///
+  /// The return value should be sent back to the client in a
+  /// `Sec-WebSocket-Accept` header.
+  ///
+  /// [initial handshake]: https://tools.ietf.org/html/rfc6455#section-4.2.2
+  static String signKey(String key) {
+    var hash = new SHA1();
+    // We use [codeUnits] here rather than UTF-8-decoding the string because
+    // [key] is expected to be base64 encoded, and so will be pure ASCII.
+    hash.add((key + _webSocketGUID).codeUnits);
+    return CryptoUtils.bytesToBase64(hash.close());
+  }
+
+  /// Creates a new WebSocket handling messaging across an existing socket.
+  ///
+  /// Because this is HTTP-API-agnostic, the initial [WebSocket handshake][]
+  /// must have already been completed on the socket before this is called.
+  ///
+  /// If [stream] is also a [StreamSink] (for example, if it's a "dart:io"
+  /// `Socket`), it will be used for both sending and receiving data. Otherwise,
+  /// it will be used for receiving data and [sink] will be used for sending it.
+  ///
+  /// If this is a WebSocket server, [serverSide] should be `true` (the
+  /// default); if it's a client, [serverSide] should be `false`.
+  ///
+  /// [WebSocket handshake]: https://tools.ietf.org/html/rfc6455#section-4
+  factory CompatibleWebSocket(Stream<List<int>> stream,
+        {StreamSink<List<int>> sink, bool serverSide: true}) {
+    if (sink == null) {
+      if (stream is! StreamSink) {
+        throw new ArgumentError("If stream isn't also a StreamSink, sink must "
+            "be passed explicitly.");
+      }
+      sink = stream as StreamSink;
+    }
+
+    return new _WebSocketImpl._fromSocket(stream, sink, serverSide);
+  }
+
+  /// Closes the web socket connection.
+  ///
+  /// [closeCode] and [closeReason] are the [close code][] and [reason][] sent
+  /// to the remote peer, respectively. If they are omitted, the peer will see
+  /// a "no status received" code with no reason.
+  ///
+  /// [close code]: https://tools.ietf.org/html/rfc6455#section-7.1.5
+  /// [reason]: https://tools.ietf.org/html/rfc6455#section-7.1.6
+  Future close([int closeCode, String closeReason]);
+}
+
+/// An exception thrown by [CompatibleWebSocket].
+class CompatibleWebSocketException implements Exception {
+  final String message;
+
+  CompatibleWebSocketException([this.message]);
+
+  String toString() => message == null
+      ? "CompatibleWebSocketException" :
+        "CompatibleWebSocketException: $message";
+}
+
+// The following code is copied from sdk/lib/io/websocket_impl.dart. The
+// "dart:io" implementation isn't used directly both to support non-"dart:io"
+// applications, and because it's incompatible with non-"dart:io" HTTP requests
+// (issue 18172).
+//
+// Because it's copied directly, only modifications necessary to support the
+// desired public API and to remove "dart:io" dependencies have been made.
+
+/**
+ * Web socket status codes used when closing a web socket connection.
+ */
+abstract class _WebSocketStatus {
+  static const int NORMAL_CLOSURE = 1000;
+  static const int GOING_AWAY = 1001;
+  static const int PROTOCOL_ERROR = 1002;
+  static const int UNSUPPORTED_DATA = 1003;
+  static const int RESERVED_1004  = 1004;
+  static const int NO_STATUS_RECEIVED = 1005;
+  static const int ABNORMAL_CLOSURE = 1006;
+  static const int INVALID_FRAME_PAYLOAD_DATA = 1007;
+  static const int POLICY_VIOLATION = 1008;
+  static const int MESSAGE_TOO_BIG = 1009;
+  static const int MISSING_MANDATORY_EXTENSION = 1010;
+  static const int INTERNAL_SERVER_ERROR = 1011;
+  static const int RESERVED_1015 = 1015;
+}
+
+abstract class _WebSocketState {
+  static const int CONNECTING = 0;
+  static const int OPEN = 1;
+  static const int CLOSING = 2;
+  static const int CLOSED = 3;
+}
 
 const String _webSocketGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
 
+final _random = new Random();
+
 // Matches _WebSocketOpcode.
 class _WebSocketMessageType {
   static const int NONE = 0;
   static const int TEXT = 1;
   static const int BINARY = 2;
 }
 
 
 class _WebSocketOpcode {
   static const int CONTINUATION = 0;
@@ skipping 36 matching lines @@
   int _state = START;
   bool _fin = false;
   int _opcode = -1;
   int _len = -1;
   bool _masked = false;
   int _remainingLenBytes = -1;
   int _remainingMaskingKeyBytes = 4;
   int _remainingPayloadBytes = -1;
   int _unmaskingIndex = 0;
   int _currentMessageType = _WebSocketMessageType.NONE;
-  int closeCode = WebSocketStatus.NO_STATUS_RECEIVED;
+  int closeCode = _WebSocketStatus.NO_STATUS_RECEIVED;
   String closeReason = "";
 
   EventSink _eventSink;
 
   final bool _serverSide;
   final List _maskingBytes = new List(4);
   final BytesBuilder _payload = new BytesBuilder(copy: false);
 
   _WebSocketProtocolTransformer([this._serverSide = false]);
 
@@ skipping 15 matching lines @@
   void close() => _eventSink.close();
 
   /**
    * Process data received from the underlying communication channel.
    */
   void add(Uint8List buffer) {
     int count = buffer.length;
     int index = 0;
     int lastIndex = count;
     if (_state == CLOSED) {
-      throw new WebSocketException("Data on closed connection");
+      throw new CompatibleWebSocketException("Data on closed connection");
     }
     if (_state == FAILURE) {
-      throw new WebSocketException("Data on failed connection");
+      throw new CompatibleWebSocketException("Data on failed connection");
     }
     while ((index < lastIndex) && _state != CLOSED && _state != FAILURE) {
       int byte = buffer[index];
       if (_state <= LEN_REST) {
         if (_state == START) {
           _fin = (byte & 0x80) != 0;
           if ((byte & 0x70) != 0) {
             // The RSV1, RSV2 bits RSV3 must be all zero.
-            throw new WebSocketException("Protocol error");
+            throw new CompatibleWebSocketException("Protocol error");
           }
           _opcode = (byte & 0xF);
           if (_opcode <= _WebSocketOpcode.BINARY) {
             if (_opcode == _WebSocketOpcode.CONTINUATION) {
               if (_currentMessageType == _WebSocketMessageType.NONE) {
-                throw new WebSocketException("Protocol error");
+                throw new CompatibleWebSocketException("Protocol error");
               }
             } else {
               assert(_opcode == _WebSocketOpcode.TEXT ||
                      _opcode == _WebSocketOpcode.BINARY);
               if (_currentMessageType != _WebSocketMessageType.NONE) {
-                throw new WebSocketException("Protocol error");
+                throw new CompatibleWebSocketException("Protocol error");
               }
               _currentMessageType = _opcode;
             }
           } else if (_opcode >= _WebSocketOpcode.CLOSE &&
                      _opcode <= _WebSocketOpcode.PONG) {
             // Control frames cannot be fragmented.
-            if (!_fin) throw new WebSocketException("Protocol error");
+            if (!_fin) throw new CompatibleWebSocketException("Protocol error");
           } else {
-            throw new WebSocketException("Protocol error");
+            throw new CompatibleWebSocketException("Protocol error");
           }
           _state = LEN_FIRST;
         } else if (_state == LEN_FIRST) {
           _masked = (byte & 0x80) != 0;
           _len = byte & 0x7F;
           if (_isControlFrame() && _len > 125) {
-            throw new WebSocketException("Protocol error");
+            throw new CompatibleWebSocketException("Protocol error");
           }
           if (_len == 126) {
             _len = 0;
             _remainingLenBytes = 2;
             _state = LEN_REST;
           } else if (_len == 127) {
             _len = 0;
             _remainingLenBytes = 8;
             _state = LEN_REST;
           } else {
@@ skipping 25 matching lines @@
           }
           // Control frame and data frame share _payloads.
           _payload.add(
               new Uint8List.view(buffer.buffer, index, payloadLength));
           index += payloadLength;
           if (_isControlFrame()) {
             if (_remainingPayloadBytes == 0) _controlFrameEnd();
           } else {
             if (_currentMessageType != _WebSocketMessageType.TEXT &&
                 _currentMessageType != _WebSocketMessageType.BINARY) {
-                throw new WebSocketException("Protocol error");
+                throw new CompatibleWebSocketException("Protocol error");
             }
             if (_remainingPayloadBytes == 0) _messageFrameEnd();
           }
 
           // Hack - as we always do index++ below.
           index--;
         }
       }
 
       // Move to the next byte.
@@ skipping 34 matching lines @@
     // Handle end.
     final int end = index + length;
     for (int i = index; i < end; i++) {
       buffer[i] ^= _maskingBytes[_unmaskingIndex++ & 3];
     }
   }
 
   void _lengthDone() {
     if (_masked) {
       if (!_serverSide) {
-        throw new WebSocketException("Received masked frame from server");
+        throw new CompatibleWebSocketException(
+            "Received masked frame from server");
       }
       _state = MASK;
     } else {
       if (_serverSide) {
-        throw new WebSocketException("Received unmasked frame from client");
+        throw new CompatibleWebSocketException(
+            "Received unmasked frame from client");
       }
       _remainingPayloadBytes = _len;
       _startPayload();
     }
   }
 
   void _maskDone() {
     _remainingPayloadBytes = _len;
     _startPayload();
   }
@@ skipping 35 matching lines @@
           break;
       }
       _currentMessageType = _WebSocketMessageType.NONE;
     }
     _prepareForNextFrame();
   }
 
   void _controlFrameEnd() {
     switch (_opcode) {
       case _WebSocketOpcode.CLOSE:
-        closeCode = WebSocketStatus.NO_STATUS_RECEIVED;
+        closeCode = _WebSocketStatus.NO_STATUS_RECEIVED;
         var payload = _payload.takeBytes();
         if (payload.length > 0) {
           if (payload.length == 1) {
-            throw new WebSocketException("Protocol error");
+            throw new CompatibleWebSocketException("Protocol error");
           }
           closeCode = payload[0] << 8 | payload[1];
-          if (closeCode == WebSocketStatus.NO_STATUS_RECEIVED) {
-            throw new WebSocketException("Protocol error");
+          if (closeCode == _WebSocketStatus.NO_STATUS_RECEIVED) {
+            throw new CompatibleWebSocketException("Protocol error");
           }
           if (payload.length > 2) {
             closeReason = UTF8.decode(payload.sublist(2));
           }
         }
         _state = CLOSED;
         _eventSink.close();
         break;
 
       case _WebSocketOpcode.PING:
@@ skipping 30 matching lines @@
   final List<int> payload;
   _WebSocketPing([this.payload = null]);
 }
 
 
 class _WebSocketPong {
   final List<int> payload;
   _WebSocketPong([this.payload = null]);
 }
 
-
-class _WebSocketTransformerImpl implements WebSocketTransformer {
-  final StreamController<WebSocket> _controller =
-      new StreamController<WebSocket>(sync: true);
-  final Function _protocolSelector;
-
-  _WebSocketTransformerImpl(this._protocolSelector);
-
-  Stream<WebSocket> bind(Stream<HttpRequest> stream) {
-    stream.listen((request) {
-        _upgrade(request, _protocolSelector)
-            .then((WebSocket webSocket) => _controller.add(webSocket))
-            .catchError(_controller.addError);
-    });
-
-    return _controller.stream;
-  }
-
-  static Future<WebSocket> _upgrade(HttpRequest request, _protocolSelector) {
-    var response = request.response;
-    if (!_isUpgradeRequest(request)) {
-      // Send error response.
-      response
-          ..statusCode = HttpStatus.BAD_REQUEST
-          ..close();
-      return new Future.error(
-          new WebSocketException("Invalid WebSocket upgrade request"));
-    }
-
-    Future upgrade(String protocol) {
-      // Send the upgrade response.
-      response
-          ..statusCode = HttpStatus.SWITCHING_PROTOCOLS
-          ..headers.add(HttpHeaders.CONNECTION, "Upgrade")
-          ..headers.add(HttpHeaders.UPGRADE, "websocket");
-      String key = request.headers.value("Sec-WebSocket-Key");
-      _SHA1 sha1 = new _SHA1();
-      sha1.add("$key$_webSocketGUID".codeUnits);
-      String accept = _CryptoUtils.bytesToBase64(sha1.close());
-      response.headers.add("Sec-WebSocket-Accept", accept);
-      if (protocol != null && protocol.isNotEmpty) {
-        response.headers.add("Sec-WebSocket-Protocol", protocol);
-      }
-      response.headers.contentLength = 0;
-      return response.detachSocket()
-          .then((socket) => new _WebSocketImpl._fromSocket(
-                socket, protocol, true));
-    }
-
-    var protocols = request.headers['Sec-WebSocket-Protocol'];
-    if (protocols != null && _protocolSelector != null) {
-      // The suggested protocols can be spread over multiple lines, each
-      // consisting of multiple protocols. To unify all of them, first join
-      // the lists with ', ' and then tokenize.
-      protocols = _HttpParser._tokenizeFieldValue(protocols.join(', '));
-      return new Future(() => _protocolSelector(protocols))
-        .then((protocol) {
-          if (protocols.indexOf(protocol) < 0) {
-            throw new WebSocketException(
-                "Selected protocol is not in the list of available protocols");
-          }
-          return protocol;
-        })
-        .catchError((error) {
-          response
-              ..statusCode = HttpStatus.INTERNAL_SERVER_ERROR
-              ..close();
-          throw error;
-        })
-        .then(upgrade);
-    } else {
-      return upgrade(null);
-    }
-  }
-
-  static bool _isUpgradeRequest(HttpRequest request) {
-    if (request.method != "GET") {
-      return false;
-    }
-    if (request.headers[HttpHeaders.CONNECTION] == null) {
-      return false;
-    }
-    bool isUpgrade = false;
-    request.headers[HttpHeaders.CONNECTION].forEach((String value) {
-      if (value.toLowerCase() == "upgrade") isUpgrade = true;
-    });
-    if (!isUpgrade) return false;
-    String upgrade = request.headers.value(HttpHeaders.UPGRADE);
-    if (upgrade == null || upgrade.toLowerCase() != "websocket") {
-      return false;
-    }
-    String version = request.headers.value("Sec-WebSocket-Version");
-    if (version == null || version != "13") {
-      return false;
-    }
-    String key = request.headers.value("Sec-WebSocket-Key");
-    if (key == null) {
-      return false;
-    }
-    return true;
-  }
-}
-
-
 // TODO(ajohnsen): Make this transformer reusable.
 class _WebSocketOutgoingTransformer implements StreamTransformer, EventSink {
   final _WebSocketImpl webSocket;
   EventSink _eventSink;
 
   _WebSocketOutgoingTransformer(this.webSocket);
 
   Stream bind(Stream stream) {
     return new Stream.eventTransformed(
         stream,
@@ skipping 79 matching lines @@
     } else if (dataLength > 125) {
       header[index++] = 126;
       lengthBytes = 2;
     }
     // Write the length in network byte order into the header.
     for (int i = 0; i < lengthBytes; i++) {
       header[index++] = dataLength >> (((lengthBytes - 1) - i) * 8) & 0xFF;
     }
     if (mask) {
       header[1] |= 1 << 7;
-      var maskBytes = _IOCrypto.getRandomBytes(4);
+      var maskBytes = [_random.nextInt(256), _random.nextInt(256),
+          _random.nextInt(256), _random.nextInt(256)];
       header.setRange(index, index + 4, maskBytes);
       index += 4;
       if (data != null) {
         Uint8List list;
         // If this is a text message just do the masking inside the
         // encoded data.
         if (opcode == _WebSocketOpcode.TEXT && data is Uint8List) {
           list = data;
         } else {
           if (data is Uint8List) {
@@ skipping 37 matching lines @@
       return [header];
     } else {
       return [header, data];
     }
   }
 }
 
 
 class _WebSocketConsumer implements StreamConsumer {
   final _WebSocketImpl webSocket;
-  final Socket socket;
+  final StreamSink<List<int>> sink;
   StreamController _controller;
   StreamSubscription _subscription;
   bool _issuedPause = false;
   bool _closed = false;
   Completer _closeCompleter = new Completer();
   Completer _completer;
 
-  _WebSocketConsumer(this.webSocket, this.socket);
+  _WebSocketConsumer(this.webSocket, this.sink);
 
   void _onListen() {
     if (_subscription != null) {
       _subscription.cancel();
     }
   }
 
   void _onPause() {
     if (_subscription != null) {
       _subscription.pause();
@@ skipping 19 matching lines @@
   }
 
   _ensureController() {
     if (_controller != null) return;
     _controller = new StreamController(sync: true,
                                        onPause: _onPause,
                                        onResume: _onResume,
                                        onCancel: _onListen);
     var stream = _controller.stream.transform(
         new _WebSocketOutgoingTransformer(webSocket));
-    socket.addStream(stream)
+    sink.addStream(stream)
         .then((_) {
           _done();
           _closeCompleter.complete(webSocket);
         }, onError: (error, StackTrace stackTrace) {
           _closed = true;
           _cancel();
           if (error is ArgumentError) {
             if (!_done(error, stackTrace)) {
               _closeCompleter.completeError(error, stackTrace);
             }
@@ skipping 32 matching lines @@
     if (_issuedPause) {
       _subscription.pause();
       _issuedPause = false;
     }
     return _completer.future;
   }
 
   Future close() {
     _ensureController();
     Future closeSocket() {
-      return socket.close().catchError((_) {}).then((_) => webSocket);
+      return sink.close().catchError((_) {}).then((_) => webSocket);
     }
     _controller.close();
     return _closeCompleter.future.then((_) => closeSocket());
   }
 
   void add(data) {
     if (_closed) return;
     _ensureController();
     _controller.add(data);
   }
 
   void closeSocket() {
     _closed = true;
     _cancel();
     close();
   }
 }
 
 
-class _WebSocketImpl extends Stream implements WebSocket {
-  final String protocol;
-
+class _WebSocketImpl extends Stream implements CompatibleWebSocket {
   StreamController _controller;
   StreamSubscription _subscription;
-  StreamSink _sink;
+  StreamController _sink;
 
-  final Socket _socket;
   final bool _serverSide;
-  int _readyState = WebSocket.CONNECTING;
+  int _readyState = _WebSocketState.CONNECTING;
   bool _writeClosed = false;
   int _closeCode;
   String _closeReason;
   Duration _pingInterval;
   Timer _pingTimer;
   _WebSocketConsumer _consumer;
 
   int _outCloseCode;
   String _outCloseReason;
   Timer _closeTimer;
 
-  static final HttpClient _httpClient = new HttpClient();
-
-  static Future<WebSocket> connect(String url, List<String> protocols) {
-    Uri uri = Uri.parse(url);
-    if (uri.scheme != "ws" && uri.scheme != "wss") {
-      throw new WebSocketException("Unsupported URL scheme '${uri.scheme}'");
-    }
-    if (uri.userInfo != "") {
-      throw new WebSocketException("Unsupported user info '${uri.userInfo}'");
-    }
-
-    Random random = new Random();
-    // Generate 16 random bytes.
-    Uint8List nonceData = new Uint8List(16);
-    for (int i = 0; i < 16; i++) {
-      nonceData[i] = random.nextInt(256);
-    }
-    String nonce = _CryptoUtils.bytesToBase64(nonceData);
-
-    uri = new Uri(scheme: uri.scheme == "wss" ? "https" : "http",
-                  userInfo: uri.userInfo,
-                  host: uri.host,
-                  port: uri.port,
-                  path: uri.path,
-                  query: uri.query,
-                  fragment: uri.fragment);
-    return _httpClient.openUrl("GET", uri)
-      .then((request) {
-        // Setup the initial handshake.
-        request.headers
-            ..add(HttpHeaders.CONNECTION, "Upgrade")
-            ..set(HttpHeaders.UPGRADE, "websocket")
-            ..set("Sec-WebSocket-Key", nonce)
-            ..set("Cache-Control", "no-cache")
-            ..set("Sec-WebSocket-Version", "13");
-        if (protocols.isNotEmpty) {
-          request.headers.add("Sec-WebSocket-Protocol", protocols);
-        }
-        return request.close();
-      })
-      .then((response) {
-        void error(String message) {
-          // Flush data.
-          response.detachSocket().then((socket) {
-            socket.destroy();
-          });
-          throw new WebSocketException(message);
-        }
-        if (response.statusCode != HttpStatus.SWITCHING_PROTOCOLS ||
-            response.headers[HttpHeaders.CONNECTION] == null ||
-            !response.headers[HttpHeaders.CONNECTION].any(
-                (value) => value.toLowerCase() == "upgrade") ||
-            response.headers.value(HttpHeaders.UPGRADE).toLowerCase() !=
-                "websocket") {
-          error("Connection to '$uri' was not upgraded to websocket");
-        }
-        String accept = response.headers.value("Sec-WebSocket-Accept");
-        if (accept == null) {
-          error("Response did not contain a 'Sec-WebSocket-Accept' header");
-        }
-        _SHA1 sha1 = new _SHA1();
-        sha1.add("$nonce$_webSocketGUID".codeUnits);
-        List<int> expectedAccept = sha1.close();
-        List<int> receivedAccept = _CryptoUtils.base64StringToBytes(accept);
-        if (expectedAccept.length != receivedAccept.length) {
-          error("Reasponse header 'Sec-WebSocket-Accept' is the wrong length");
-        }
-        for (int i = 0; i < expectedAccept.length; i++) {
-          if (expectedAccept[i] != receivedAccept[i]) {
-            error("Bad response 'Sec-WebSocket-Accept' header");
-          }
-        }
-        var protocol = response.headers.value('Sec-WebSocket-Protocol');
-        return response.detachSocket()
-            .then((socket) => new _WebSocketImpl._fromSocket(socket, protocol));
-      });
-  }
-
-  _WebSocketImpl._fromSocket(this._socket, this.protocol,
-                             [this._serverSide = false]) {
-    _consumer = new _WebSocketConsumer(this, _socket);
-    _sink = new _StreamSinkImpl(_consumer);
-    _readyState = WebSocket.OPEN;
+  _WebSocketImpl._fromSocket(Stream<List<int>> stream,
+      StreamSink<List<int>> sink, [this._serverSide = false]) {
+    _consumer = new _WebSocketConsumer(this, sink);
+    _sink = new StreamController();
+    _sink.stream.pipe(_consumer);
+    _readyState = _WebSocketState.OPEN;
 
     var transformer = new _WebSocketProtocolTransformer(_serverSide);
-    _subscription = _socket.transform(transformer).listen(
+    _subscription = stream.transform(transformer).listen(
         (data) {
           if (data is _WebSocketPing) {
             if (!_writeClosed) _consumer.add(new _WebSocketPong(data.payload));
           } else if (data is _WebSocketPong) {
             // Simply set pingInterval, as it'll cancel any timers.
             pingInterval = _pingInterval;
           } else {
             _controller.add(data);
           }
         },
         onError: (error) {
           if (_closeTimer != null) _closeTimer.cancel();
           if (error is FormatException) {
-            _close(WebSocketStatus.INVALID_FRAME_PAYLOAD_DATA);
+            _close(_WebSocketStatus.INVALID_FRAME_PAYLOAD_DATA);
           } else {
-            _close(WebSocketStatus.PROTOCOL_ERROR);
+            _close(_WebSocketStatus.PROTOCOL_ERROR);
           }
           _controller.close();
         },
         onDone: () {
           if (_closeTimer != null) _closeTimer.cancel();
-          if (_readyState == WebSocket.OPEN) {
-            _readyState = WebSocket.CLOSING;
+          if (_readyState == _WebSocketState.OPEN) {
+            _readyState = _WebSocketState.CLOSING;
             if (!_isReservedStatusCode(transformer.closeCode)) {
               _close(transformer.closeCode);
             } else {
               _close();
             }
-            _readyState = WebSocket.CLOSED;
+            _readyState = _WebSocketState.CLOSED;
           }
           _closeCode = transformer.closeCode;
           _closeReason = transformer.closeReason;
           _controller.close();
         },
         cancelOnError: true);
     _subscription.pause();
     _controller = new StreamController(sync: true,
                                        onListen: _subscription.resume,
                                        onPause: _subscription.pause,
@@ skipping 17 matching lines @@
     if (_pingTimer != null) _pingTimer.cancel();
     _pingInterval = interval;
 
     if (_pingInterval == null) return;
 
     _pingTimer = new Timer(_pingInterval, () {
       if (_writeClosed) return;
       _consumer.add(new _WebSocketPing());
       _pingTimer = new Timer(_pingInterval, () {
         // No pong received.
-        _close(WebSocketStatus.GOING_AWAY);
+        _close(_WebSocketStatus.GOING_AWAY);
       });
     });
   }
 
-  int get readyState => _readyState;
-
-  String get extensions => null;
   int get closeCode => _closeCode;
   String get closeReason => _closeReason;
 
   void add(data) => _sink.add(data);
   void addError(error, [StackTrace stackTrace]) =>
       _sink.addError(error, stackTrace);
   Future addStream(Stream stream) => _sink.addStream(stream);
   Future get done => _sink.done;
 
   Future close([int code, String reason]) {
     if (_isReservedStatusCode(code)) {
-      throw new WebSocketException("Reserved status code $code");
+      throw new CompatibleWebSocketException("Reserved status code $code");
     }
     if (_outCloseCode == null) {
       _outCloseCode = code;
       _outCloseReason = reason;
     }
     if (_closeTimer == null && !_controller.isClosed) {
       // When closing the web-socket, we no longer accept data.
       _closeTimer = new Timer(const Duration(seconds: 5), () {
         _subscription.cancel();
         _controller.close();
       });
     }
     return _sink.close();
   }
 
   void _close([int code, String reason]) {
     if (_writeClosed) return;
     if (_outCloseCode == null) {
       _outCloseCode = code;
       _outCloseReason = reason;
     }
     _writeClosed = true;
     _consumer.closeSocket();
   }
 
   static bool _isReservedStatusCode(int code) {
     return code != null &&
-           (code < WebSocketStatus.NORMAL_CLOSURE ||
-            code == WebSocketStatus.RESERVED_1004 ||
-            code == WebSocketStatus.NO_STATUS_RECEIVED ||
-            code == WebSocketStatus.ABNORMAL_CLOSURE ||
-            (code > WebSocketStatus.INTERNAL_SERVER_ERROR &&
-             code < WebSocketStatus.RESERVED_1015) ||
-            (code >= WebSocketStatus.RESERVED_1015 &&
+           (code < _WebSocketStatus.NORMAL_CLOSURE ||
+            code == _WebSocketStatus.RESERVED_1004 ||
+            code == _WebSocketStatus.NO_STATUS_RECEIVED ||
+            code == _WebSocketStatus.ABNORMAL_CLOSURE ||
+            (code > _WebSocketStatus.INTERNAL_SERVER_ERROR &&
+             code < _WebSocketStatus.RESERVED_1015) ||
+            (code >= _WebSocketStatus.RESERVED_1015 &&
              code < 3000));
   }
 }
+