Add ability to send a ping interval on a 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 437 matching lines @@
   final _WebSocketImpl webSocket;
 
   _WebSocketOutgoingTransformer(_WebSocketImpl this.webSocket);
 
   void handleData(message, EventSink<List<int>> sink) {
     if (message is _WebSocketPong) {
       addFrame(_WebSocketOpcode.PONG, message.payload, sink);
       return;
     }
     if (message is _WebSocketPing) {
-      addFrame(_WebSocketOpcode.PONG, message.payload, sink);
+      addFrame(_WebSocketOpcode.PING, message.payload, sink);
       return;
     }
     List<int> data;
     int opcode;
     if (message != null) {
       if (message is String) {
         opcode = _WebSocketOpcode.TEXT;
         data = _encodeString(message);
       } else {
         if (message is !List<int>) {
@@ skipping 95 matching lines @@
     }
   }
 }
 
 
 class _WebSocketConsumer implements StreamConsumer {
   final _WebSocketImpl webSocket;
   final Socket socket;
   StreamController _controller;
   StreamSubscription _subscription;
+  bool _issuedPause = 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;
+    }
+  }
+
   _ensureController() {
     if (_controller != null) return;
     _controller = new StreamController(sync: true,
-                                       onPause: () => _subscription.pause(),
-                                       onResume: () => _subscription.resume(),
+                                       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 (!_done(error)) {
@@ skipping 20 matching lines @@
         (data) {
           _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);
     }
     _controller.close();
     return _closeCompleter.future.then((_) => closeSocket());
   }
 
   void add(data) {
     _ensureController();
     _controller.add(data);
   }
 }
 
 
 class _WebSocketImpl extends Stream implements WebSocket {
   final StreamController _controller = new StreamController(sync: true);
   StreamSink _sink;
 
   final Socket _socket;
   final bool _serverSide;
   int _readyState = WebSocket.CONNECTING;
   bool _writeClosed = false;
   int _closeCode;
   String _closeReason;
+  Duration _pingInterval;
+  Timer _pingTimer;
+  _WebSocketConsumer _consumer;
 
   int _outCloseCode;
   String _outCloseReason;
 
   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}'");
@@ skipping 58 matching lines @@
             error("Bad response 'Sec-WebSocket-Accept' header");
           }
         }
         return response.detachSocket()
             .then((socket) => new _WebSocketImpl._fromSocket(socket));
       });
   }
 
   _WebSocketImpl._fromSocket(Socket this._socket,
                              [bool this._serverSide = false]) {
-    var consumer = new _WebSocketConsumer(this, _socket);
-    _sink = new _StreamSinkImpl(consumer);
+    _consumer = new _WebSocketConsumer(this, _socket);
+    _sink = new _StreamSinkImpl(_consumer);
     _readyState = WebSocket.OPEN;
 
     var transformer = new _WebSocketProtocolTransformer(_serverSide);
     _socket.transform(transformer).listen(
         (data) {
           if (data is _WebSocketPing) {
-            consumer.add(new _WebSocketPong(data.payload));
+            if (!_writeClosed) _consumer.add(new _WebSocketPong(data.payload));
           } else if (data is _WebSocketPong) {
-            // TODO(ajohnsen): Notify pong?
+            // 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);
           } else {
             close(WebSocketStatus.PROTOCOL_ERROR);
           }
@@ skipping 20 matching lines @@
   StreamSubscription listen(void onData(message),
                             {void onError(error),
                              void onDone(),
                              bool cancelOnError}) {
     return _controller.stream.listen(onData,
                                      onError: onError,
                                      onDone: onDone,
                                      cancelOnError: cancelOnError);
   }
 
+  Duration get pingnterval => _pingInterval;
+
+  void set pingInterval(Duration interval) {
+    if (_writeClosed) return;
+    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);
+      });
+    });
+  }
+
   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);
@@ skipping 16 matching lines @@
            (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));
   }
 }