[dart:io] webSocket pingInterval doesn't close the connection

pingInterval api pages says, when pong is not received on time, the connection will be closed. But our implementation didn't follow the description.

Bug: https://github.com/dart-lang/sdk/issues/39469
Change-Id: Ic62b8ab0152997ac60b1b836d6a1587d9ce38131
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/138760
Reviewed-by: Siva Annamalai <asiva@google.com>
Reviewed-by: Lasse R.H. Nielsen <lrn@google.com>
Commit-Queue: Zichang Guo <zichangguo@google.com>

sdk/lib/_http/websocket.dart

@@ -331,11 +331,14 @@ abstract class WebSocket
   static const int CLOSED = closed;
 
   /**
-   * Set and get 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.goingAway] close code. When a ping signal is sent, the
-   * pong message must be received within [pingInterval].
+   * The interval between ping signals.
+   *
+   * A ping message is sent every [pingInterval], starting at the first
+   * [pingInterval] after a new value has been assigned or a pong message has
+   * been received. 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.goingAway] 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.

sdk/lib/_http/websocket_impl.dart

@@ -1205,8 +1205,12 @@ class _WebSocketImpl extends Stream with _ServiceObject implements WebSocket {
       if (_writeClosed) return;
       _consumer.add(new _WebSocketPing());
       _pingTimer = new Timer(_pingInterval, () {
+        _closeTimer?.cancel();
         // No pong received.
         _close(WebSocketStatus.goingAway);
+        _closeCode = _outCloseCode;
+        _closeReason = _outCloseReason;
+        _controller.close();
       });
     });
   }

sdk_nnbd/lib/_http/websocket.dart

@@ -326,11 +326,14 @@ abstract class WebSocket
   static const int CLOSED = closed;
 
   /**
-   * Set and get 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.goingAway] close code. When a ping signal is sent, the
-   * pong message must be received within [pingInterval].
+   * The interval between ping signals.
+   *
+   * A ping message is sent every [pingInterval], starting at the first
+   * [pingInterval] after a new value has been assigned or a pong message has
+   * been received. 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.goingAway] 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.

sdk_nnbd/lib/_http/websocket_impl.dart

@@ -1203,8 +1203,12 @@ class _WebSocketImpl extends Stream with _ServiceObject implements WebSocket {
       if (_writeClosed) return;
       _consumer.add(new _WebSocketPing());
       _pingTimer = new Timer(interval, () {
+        _closeTimer?.cancel();
         // No pong received.
         _close(WebSocketStatus.goingAway);
+        _closeCode = _outCloseCode;
+        _closeReason = _outCloseReason;
+        _controller.close();
       });
     });
   }

tests/standalone/io/web_socket_ping_test.dart

@@ -36,20 +36,22 @@ void testPing(int totalConnections) {
       response.headers.add("Sec-WebSocket-Accept", accept);
       response.headers.contentLength = 0;
       response.detachSocket().then((socket) {
-        socket.drain().then((_) {
-          socket.close();
-          closed++;
-          if (closed == totalConnections) {
-            server.close();
-          }
-        });
+        socket.destroy();
       });
     });
 
+    int closeCount = 0;
     for (int i = 0; i < totalConnections; i++) {
       WebSocket.connect('ws://localhost:${server.port}').then((webSocket) {
         webSocket.pingInterval = const Duration(milliseconds: 100);
-        webSocket.drain();
+        webSocket.listen((message) {
+          Expect.fail("unexpected message");
+        }, onDone: () {
+          closeCount++;
+          if (closeCount == totalConnections) {
+            server.close();
+          }
+        });
       });
     }
   });

tests/standalone_2/io/web_socket_ping_test.dart

@@ -36,20 +36,22 @@ void testPing(int totalConnections) {
       response.headers.add("Sec-WebSocket-Accept", accept);
       response.headers.contentLength = 0;
       response.detachSocket().then((socket) {
-        socket.drain().then((_) {
-          socket.close();
-          closed++;
-          if (closed == totalConnections) {
-            server.close();
-          }
-        });
+        socket.destroy();
       });
     });
 
+    int closeCount = 0;
     for (int i = 0; i < totalConnections; i++) {
       WebSocket.connect('ws://localhost:${server.port}').then((webSocket) {
         webSocket.pingInterval = const Duration(milliseconds: 100);
-        webSocket.drain();
+        webSocket.listen((message) {
+          Expect.fail("unexpected message");
+        }, onDone: () {
+          closeCount++;
+          if (closeCount == totalConnections) {
+            server.close();
+          }
+        });
       });
     }
   });