Don't throw SocketExceptions in websocket.

sdk/lib/io/websocket_impl.dart

 // 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;
@@ skipping 554 matching lines @@
   }
 }
 
 
 class _WebSocketConsumer implements StreamConsumer {
   final _WebSocketImpl webSocket;
   final Socket socket;
   StreamController _controller;
   StreamSubscription _subscription;
   bool _issuedPause = false;
-  // Only report error if the last message was a user-provided message and not a
-  // ping or pong message.
-  bool _reportError = false;
+  bool _closed = false;
   Completer _closeCompleter = new Completer();
   Completer _completer;
 
   _WebSocketConsumer(_WebSocketImpl this.webSocket, Socket this.socket);
 
   void _onListen() {
     if (_subscription != null) {
       _subscription.cancel();
     }
   }
 
   void _onPause() {
     if (_subscription != null) {
       _subscription.pause();
     } else {
       _issuedPause = true;
     }
   }
 
   void _onResume() {
     if (_subscription != null) {
       _subscription.resume();
     } else {
       _issuedPause = false;
     }
   }
 
+  void _cancel() {
+    if (_subscription != null) {
+      var subscription = _subscription;
+      _subscription = null;
+      subscription.cancel();
+    }
+  }
+
   _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)
         .then((_) {
-                _done();
-                _closeCompleter.complete(webSocket);
-              },
-              onError: (error) {
-                if (_reportError) {
-                  if (!_done(error)) {
-                    _closeCompleter.completeError(error);
-                  }
-                } else {
-                  _done();
-                  _closeCompleter.complete(webSocket);
-                }
-              });
+          _done();
+          _closeCompleter.complete(webSocket);
+        }, onError: (error) {
+          _closed = true;
+          _cancel();
+          if (error is ArgumentError) {
+            if (!_done(error)) {
+              _closeCompleter.completeError(error);
+            }
+          } else {
+            _done();
+            _closeCompleter.complete(webSocket);
+          }
+        });
   }
 
   bool _done([error]) {
     if (_completer == null) return false;
     if (error != null) {
       _completer.completeError(error);
     } else {
       _completer.complete(webSocket);
     }
     _completer = null;
     return true;
   }
 
   Future addStream(var stream) {
+    if (_closed) {
+      stream.listen(null).cancel();
+      return new Future.value(webSocket);
+    }
     _ensureController();
     _completer = new Completer();
     _subscription = stream.listen(
         (data) {
-          _reportError = true;
           _controller.add(data);
         },
         onDone: () {
           _done();
         },
         onError: (error) {
           _done(error);
         },
         cancelOnError: true);
     if (_issuedPause) {
       _subscription.pause();
       _issuedPause = false;
     }
     return _completer.future;
   }
 
   Future close() {
     _ensureController();
     Future closeSocket() {
-      return socket.close().then((_) => webSocket);
+      return socket.close().catchError((_) {}).then((_) => webSocket);
     }
     _controller.close();
     return _closeCompleter.future.then((_) => closeSocket());
   }
 
   void add(data) {
+    if (_closed) return;
     _ensureController();
-    _reportError = false;
     _controller.add(data);
   }
+
+  void closeSocket() {
+    _closed = true;
+    _cancel();
+    close();
+  }
 }
 
 
 class _WebSocketImpl extends Stream implements WebSocket {
   StreamController _controller;
   StreamSubscription _subscription;
   StreamSink _sink;
 
   final Socket _socket;
   final bool _serverSide;
@@ skipping 92 matching lines @@
           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 (error is ArgumentError) {
-            close(WebSocketStatus.INVALID_FRAME_PAYLOAD_DATA);
+          if (error is FormatException) {
+            _close(WebSocketStatus.INVALID_FRAME_PAYLOAD_DATA);
           } else {
-            close(WebSocketStatus.PROTOCOL_ERROR);
+            _close(WebSocketStatus.PROTOCOL_ERROR);
           }
-          _controller.addError(error);
           _controller.close();
         },
         onDone: () {
           if (_readyState == WebSocket.OPEN) {
             _readyState = WebSocket.CLOSING;
             if (!_isReservedStatusCode(transformer.closeCode)) {
-              close(transformer.closeCode);
+              _close(transformer.closeCode);
             } else {
-              close();
+              _close();
             }
             _readyState = WebSocket.CLOSED;
           }
           _closeCode = transformer.closeCode;
           _closeReason = transformer.closeReason;
           _controller.close();
         },
         cancelOnError: true);
     _subscription.pause();
     _controller = new StreamController(sync: true,
@@ skipping 19 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;
   String get protocol => null;
   int get closeCode => _closeCode;
   String get closeReason => _closeReason;
 
   void add(data) => _sink.add(data);
   void addError(error) => _sink.addError(error);
   Future addStream(Stream stream) => _sink.addStream(stream);
   Future get done => _sink.done;
 
   Future close([int code, String reason]) {
-    if (!_writeClosed) {
-      if (_isReservedStatusCode(code)) {
-        throw new WebSocketException("Reserved status code $code");
-      }
+    if (_isReservedStatusCode(code)) {
+      throw new WebSocketException("Reserved status code $code");
+    }
+    if (_outCloseCode == null) {
       _outCloseCode = code;
       _outCloseReason = reason;
-      _writeClosed = true;
     }
-    if (!(_sink as _StreamSinkImpl)._isBound) _sink.close();
-    return _sink.done;
+    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 < 3000));
   }
 }