Change the web socket interface

sdk/lib/io/websocket_impl.dart

-// Copyright (c) 2012, the Dart project authors.  Please see the AUTHORS file
+// Copyright (c) 2013, 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;
 
 const String _webSocketGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
 
 class _WebSocketMessageType {
   static const int NONE = 0;
   static const int BINARY = 1;
@@ skipping 406 matching lines @@
     }
     String key = request.headers.value("Sec-WebSocket-Key");
     if (key == null) {
       return false;
     }
     return true;
   }
 }
 
 
-class _WebSocketImpl extends Stream<Event> implements WebSocket {
-  final StreamController<Event> _controller = new StreamController<Event>();
+class _WebSocketImpl extends Stream implements WebSocket {
+  final StreamController _controller = new StreamController();
 
   final _WebSocketProtocolProcessor _processor =
       new _WebSocketProtocolProcessor();
 
   final Socket _socket;
   int _readyState = WebSocket.CONNECTING;
   bool _writeClosed = false;
+  int _closeCode;
+  String _closeReason;
 
   static final HttpClient _httpClient = new HttpClient();
 
   static Future<WebSocket> connect(String url, [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}'");
@@ skipping 75 matching lines @@
     };
     _processor.onMessageData = (buffer, offset, count) {
       if (type == _WebSocketMessageType.TEXT) {
         data.add(_decodeString(buffer.getRange(offset, count)));
       } else {
         data.addAll(buffer.getRange(offset, count));
       }
     };
     _processor.onMessageEnd = () {
       if (type == _WebSocketMessageType.TEXT) {
-        _controller.add(new _WebSocketMessageEvent(data.toString()));
+        _controller.add(data.toString());
       } else {
-        _controller.add(new _WebSocketMessageEvent(data));
+        _controller.add(data);
       }
     };
     _processor.onClosed = (code, reason) {
       bool clean = true;
       if (_readyState == WebSocket.OPEN) {
         _readyState = WebSocket.CLOSING;
         if (code != WebSocketStatus.NO_STATUS_RECEIVED) {
           _close(code);
         } else {
           _close();
           clean = false;
         }
         _readyState = WebSocket.CLOSED;
       }
       if (_readyState == WebSocket.CLOSED) return;
-      _controller.add(new _WebSocketCloseEvent(clean, code, reason));
+      //_cleanClose = clean;

[Anders Johnsen, 2013/03/04 10:29:29]

Outcommented code.

[Søren Gjesse, 2013/03/04 10:35:18]

Oops - removed.

+      _closeCode = code;
+      _closeReason = reason;
       _controller.close();
     };
 
     _socket.listen(
         (data) => _processor.update(data, 0, data.length),
         onDone: () => _processor.closed(),
         onError: (error) => _controller.signalError(error));
 
     _socket.done
         .catchError((error) {
           if (_readyState == WebSocket.CLOSED) return;
           _readyState = WebSocket.CLOSED;
+          //_cleanClose = false;

[Anders Johnsen, 2013/03/04 10:29:29]

Ditto.

[Søren Gjesse, 2013/03/04 10:35:18]

Ditto.

+          _closeCode = ABNORMAL_CLOSURE;
           _controller.signalError(error);
           _controller.close();
           _socket.destroy();
         })
         .whenComplete(() {
           _writeClosed = true;
         });
   }
 
-  StreamSubscription<Event> listen(void onData(Event event),
-                                   {void onError(AsyncError error),
-                                    void onDone(),
-                                    bool unsubscribeOnError}) {
+  StreamSubscription listen(void onData(message),
+                            {void onError(AsyncError error),
+                             void onDone(),
+                             bool unsubscribeOnError}) {
     return _controller.stream.listen(onData,
                                      onError: onError,
                                      onDone: onDone,
                                      unsubscribeOnError: unsubscribeOnError);
   }
 
   int get readyState => _readyState;
-  int get bufferedAmount => 0;
 
   String get extensions => null;
   String get protocol => null;
+  int get closeCode => _closeCode;
+  String get closeReason => _closeReason;
 
   void close([int code, String reason]) {
     if (_readyState < WebSocket.CLOSING) _readyState = WebSocket.CLOSING;
     if (code == WebSocketStatus.RESERVED_1004 ||
         code == WebSocketStatus.NO_STATUS_RECEIVED ||
         code == WebSocketStatus.RESERVED_1015) {
       throw new WebSocketException("Reserved status code $code");
     }
     _close(code, reason);
   }
@@ skipping 81 matching lines @@
     for (int i = 0; i < lengthBytes; i++) {
       header[index++] = dataLength >> (((lengthBytes - 1) - i) * 8) & 0xFF;
     }
     assert(index == headerSize);
     _socket.add(header);
     if (data != null) {
       _socket.add(data);
     }
   }
 }
-
-
-class _WebSocketMessageEvent implements MessageEvent {
-  _WebSocketMessageEvent(this._data);
-  get data => _data;
-  var _data;
-}
-
-
-class _WebSocketCloseEvent implements CloseEvent {
-  _WebSocketCloseEvent(this._wasClean, this._code, this._reason);
-  bool get wasClean => _wasClean;
-  int get code => _code;
-  String get reason => _reason;
-  bool _wasClean;
-  int _code;
-  String _reason;
-}