Implement correct back-throttleing for web-sockets.

sdk/lib/io/websocket_impl.dart

Index: sdk/lib/io/websocket_impl.dart
diff --git a/sdk/lib/io/websocket_impl.dart b/sdk/lib/io/websocket_impl.dart
index c054b1ec9e0bc301a1004a05f04052951e6be5a3..45b1f24f771a38cafd396726d42fadba4e181689 100644
--- a/sdk/lib/io/websocket_impl.dart
+++ b/sdk/lib/io/websocket_impl.dart
@@ -572,6 +572,7 @@ class _WebSocketConsumer implements StreamConsumer {
   StreamController _controller;
   StreamSubscription _subscription;
   bool _issuedPause = false;
+  bool _reportError = false;
   Completer _closeCompleter = new Completer();
   Completer _completer;
 
@@ -613,8 +614,13 @@ class _WebSocketConsumer implements StreamConsumer {
                 _closeCompleter.complete(webSocket);
               },
               onError: (error) {
-                if (!_done(error)) {
-                  _closeCompleter.completeError(error);
+                if (_reportError) {
+                  if (!_done(error)) {
+                    _closeCompleter.completeError(error);
+                  }
+                } else {
+                  _done();
+                  _closeCompleter.complete(webSocket);
                 }
               });
   }
@@ -635,6 +641,7 @@ class _WebSocketConsumer implements StreamConsumer {
     _completer = new Completer();
     _subscription = stream.listen(
         (data) {
+          _reportError = true;
           _controller.add(data);
         },
         onDone: () {
@@ -662,13 +669,14 @@ class _WebSocketConsumer implements StreamConsumer {
 
   void add(data) {
     _ensureController();
+    _reportError = false;
     _controller.add(data);
   }
 }
 
 
 class _WebSocketImpl extends Stream implements WebSocket {
-  final StreamController _controller = new StreamController(sync: true);
+  StreamController _controller;
   StreamSink _sink;
 
   final Socket _socket;
@@ -680,6 +688,7 @@ class _WebSocketImpl extends Stream implements WebSocket {
   Duration _pingInterval;
   Timer _pingTimer;
   _WebSocketConsumer _consumer;
+  var _subscription;
 
   int _outCloseCode;
   String _outCloseReason;
@@ -763,13 +772,14 @@ class _WebSocketImpl extends Stream implements WebSocket {
     _readyState = WebSocket.OPEN;
 
     var transformer = new _WebSocketProtocolTransformer(_serverSide);
-    _socket.transform(transformer).listen(
+    _subscription = _socket.transform(transformer).listen(
         (data) {
+          // Simply set pingInterval, as it'll cancel any timers.
+          pingInterval = _pingInterval;
           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;
+            // Already handled.
           } else {
             _controller.add(data);
           }
@@ -798,6 +808,11 @@ class _WebSocketImpl extends Stream implements WebSocket {
           _controller.close();
         },
         cancelOnError: true);
+    _subscription.pause();
+    _controller = new StreamController(sync: true,
+                                       onListen: _subscription.resume,
+                                       onPause: _subscription.pause,
+                                       onResume: _subscription.resume);
   }
 
   StreamSubscription listen(void onData(message),
@@ -824,6 +839,8 @@ class _WebSocketImpl extends Stream implements WebSocket {
       _consumer.add(new _WebSocketPing());
       _pingTimer = new Timer(_pingInterval, () {
         // No pong received.
+        _closeCode = WebSocketStatus.GOING_AWAY;
+        _subscription.cancel();
         close(WebSocketStatus.GOING_AWAY);
       });
     });