Fix flaky peer connection (#36089)

Fixes https://github.com/flutter/flutter/issues/36091.

Previously, a sendRequest will be sent even if the peer is closed during
a driver test. That will cause a time out without any error information.
Such issue is unreproducible on my Mac Book Pro, but 100% reproducible
on our new Mac mini (2018).

The closing issue is tracked in https://github.com/flutter/flutter/issues/36268

Additional to this fix, we should also patch the jason_rpc_2 so the peer
will throw exception if sendRequest is attempted while the connection is
closed.

**Test**:
tiles_scroll_perf_iphonexs__timeline_summary failed without this patch.
It will pass after this patch.

I'm not sure how to add a unit test for this. Please let me know if you have
some ideas.

This patch will generate the following warning log on the new Mac Mini
```
flutter: Observatory listening on http://127.0.0.1:50192/cZPDF4sW7MM=/  
Installing and launching...                                        10.2s
00:00 +0: scrolling performance test (setUpAll)
[info ] FlutterDriver: Connecting to Flutter application at http://127.0.0.1:1069/cZPDF4sW7MM=/
[trace] FlutterDriver: Isolate found with number: 3684677742843303
[trace] FlutterDriver: Isolate is paused at start.
[trace] FlutterDriver: Attempting to resume isolate
[trace] FlutterDriver: Waiting for service extension
[info ] FlutterDriver: Connected to Flutter application.
00:00 +0: scrolling performance test complex_layout_scroll_perf
[warning] FlutterDriver: Instance of '_WebSocketImpl' is closed with an unexpected code 1005
[warning] FlutterDriver: Peer connection is closed! Trying to restore the connection...
00:10 +1: scrolling performance test tiles_scroll_perf
[warning] FlutterDriver: Instance of '_WebSocketImpl' is closed with an unexpected code 1005
[warning] FlutterDriver: Peer connection is closed! Trying to restore the connection...
00:20 +2: scrolling performance test (tearDownAll)
00:20 +2: All tests passed!
Stopping application instance.
```

packages/flutter_driver/lib/src/driver/driver.dart

@@ -268,6 +268,8 @@ class FlutterDriver {
       logCommunicationToFile: logCommunicationToFile,
     );
 
+    driver._dartVmReconnectUrl = dartVmServiceUrl;
+
     // Attempts to resume the isolate, but does not crash if it fails because
     // the isolate is already resumed. There could be a race with other tools,
     // such as a debugger, any of which could have resumed the isolate.
@@ -392,7 +394,27 @@ class FlutterDriver {
   final VMServiceClient serviceClient;
 
   /// JSON-RPC client useful for sending raw JSON requests.
-  final rpc.Peer _peer;
+  rpc.Peer _peer;
+
+  String _dartVmReconnectUrl;
+
+  Future<void> _restorePeerConnectionIfNeeded() async {
+    if (!_peer.isClosed || _dartVmReconnectUrl == null) {
+      return;
+    }
+
+    _log.warning(
+        'Peer connection is closed! Trying to restore the connection...'
+    );
+
+    final String webSocketUrl = _getWebSocketUrl(_dartVmReconnectUrl);
+    final WebSocket ws = await WebSocket.connect(webSocketUrl);
+    ws.done.then((dynamic _) => _checkCloseCode(ws));
+    _peer = rpc.Peer(
+        IOWebSocketChannel(ws).cast(),
+        onUnhandledError: _unhandledJsonRpcError,
+    )..listen();
+  }
 
   /// The main isolate hosting the Flutter application.
   ///
@@ -840,6 +862,7 @@ class FlutterDriver {
   ///
   /// [getFlagList]: https://github.com/dart-lang/sdk/blob/master/runtime/vm/service/service.md#getflaglist
   Future<List<Map<String, dynamic>>> getVmFlags() async {
+    await _restorePeerConnectionIfNeeded();
     final Map<String, dynamic> result = await _peer.sendRequest('getFlagList');
     return result != null
         ? result['flags'].cast<Map<String,dynamic>>()
@@ -1084,9 +1107,7 @@ void _unhandledJsonRpcError(dynamic error, dynamic stack) {
   // assert(false);
 }
 
-/// Waits for a real Dart VM service to become available, then connects using
-/// the [VMServiceClient].
-Future<VMServiceClientConnection> _waitAndConnect(String url) async {
+String _getWebSocketUrl(String url) {
   Uri uri = Uri.parse(url);
   final List<String> pathSegments = <String>[];
   // If there's an authentication code (default), we need to add it to our path.
@@ -1096,16 +1117,36 @@ Future<VMServiceClientConnection> _waitAndConnect(String url) async {
   pathSegments.add('ws');
   if (uri.scheme == 'http')
     uri = uri.replace(scheme: 'ws', pathSegments: pathSegments);
+  return uri.toString();
+}
+
+void _checkCloseCode(WebSocket ws) {
+  if (ws.closeCode != 1000 && ws.closeCode != null) {
+    _log.warning('$ws is closed with an unexpected code ${ws.closeCode}');
+  }
+}
+
+/// Waits for a real Dart VM service to become available, then connects using
+/// the [VMServiceClient].
+Future<VMServiceClientConnection> _waitAndConnect(String url) async {
+  final String webSocketUrl = _getWebSocketUrl(url);
   int attempts = 0;
   while (true) {
     WebSocket ws1;
     WebSocket ws2;
     try {
-      ws1 = await WebSocket.connect(uri.toString());
-      ws2 = await WebSocket.connect(uri.toString());
+      ws1 = await WebSocket.connect(webSocketUrl);
+      ws2 = await WebSocket.connect(webSocketUrl);
+
+      ws1.done.then((dynamic _) => _checkCloseCode(ws1));
+      ws2.done.then((dynamic _) => _checkCloseCode(ws2));
+
       return VMServiceClientConnection(
         VMServiceClient(IOWebSocketChannel(ws1).cast()),
-        rpc.Peer(IOWebSocketChannel(ws2).cast(), onUnhandledError: _unhandledJsonRpcError)..listen(),
+        rpc.Peer(
+            IOWebSocketChannel(ws2).cast(),
+            onUnhandledError: _unhandledJsonRpcError,
+        )..listen(),
       );
     } catch (e) {
       await ws1?.close();

packages/flutter_driver/test/flutter_driver_test.dart

@@ -599,4 +599,7 @@ class MockVMPauseBreakpointEvent extends Mock implements VMPauseBreakpointEvent
 
 class MockVMResumeEvent extends Mock implements VMResumeEvent { }
 
-class MockPeer extends Mock implements rpc.Peer { }
+class MockPeer extends Mock implements rpc.Peer {
+  @override
+  bool get isClosed => false;
+}