Properly handle web socket requests to half-initialized servers.

sdk/lib/_internal/pub/test/serve/utils.dart

 // Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS d.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.
 
 library pub_tests;
 
 import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
 
@@ skipping 275 matching lines @@
   // Use the existing one if already connected.
   if (_webSocket != null) return new Future.value();
 
   // Server should already be running.
   expect(_pubServer, isNotNull);
   expect(_adminPort, isNotNull);
 
   return WebSocket.connect("ws://127.0.0.1:$_adminPort").then((socket) {
     _webSocket = socket;
     // TODO(rnystrom): Works around #13913.
-    _webSocketBroadcastStream = _webSocket.asBroadcastStream();
+    _webSocketBroadcastStream = _webSocket.map(JSON.decode).asBroadcastStream();
   });
 }
 
 /// Sends a JSON RPC 2.0 request to the running pub serve's web socket
+/// connection.
+///
+/// This calls a method named [method] with the given [params]. [params] may
+/// contain Futures, in which case this will wait until they've completed before
+/// sending the request.
+///
+/// This schedules the request, but doesn't block the schedule on the response.
+/// It returns the response as a [Future].
+Future<Map> webSocketRequest(String method, Map params) {
+  var completer = new Completer();
+  schedule(() {
+    return Future.wait([
+      _ensureWebSocket(),
+      awaitObject(params),
+    ]).then((results) {
+      var resolvedParams = results[1];
+      chainToCompleter(
+          currentSchedule.wrapFuture(_jsonRpcRequest(method, resolvedParams)),
+          completer);
+    });
+  }, "send $method with $params to web socket");
+  return completer.future;
+}
+
+/// Sends a JSON RPC 2.0 request to the running pub serve's web socket
 /// connection, waits for a reply, then verifies the result.
 ///
 /// This calls a method named [method] with the given [params]. [params] may
 /// contain Futures, in which case this will wait until they've completed before
 /// sending the request.
 ///
 /// The result is validated using [result], which may be a [Matcher] or a [Map]
 /// containing [Matcher]s and [Future]s. This will wait until any futures are
 /// completed before sending the request.
 ///
 /// Returns a [Future] that completes to the call's result.
 Future<Map> expectWebSocketResult(String method, Map params, result) {
   return schedule(() {
     return Future.wait([
-      _ensureWebSocket(),
-      awaitObject(params),
+      webSocketRequest(method, params),
       awaitObject(result)
     ]).then((results) {
-      var resolvedParams = results[1];
-      var resolvedResult = results[2];
-
-      return _jsonRpcRequest(method, resolvedParams).then((response) {
-        expect(response["result"], resolvedResult);
-        return response["result"];
-      });
+      var response = results[0];
+      var resolvedResult = results[1];
+      expect(response["result"], resolvedResult);
+      return response["result"];
     });
   }, "send $method with $params to web socket and expect $result");
 }
 
 /// Sends a JSON RPC 2.0 request to the running pub serve's web socket
 /// connection, waits for a reply, then verifies the error response.
 ///
 /// This calls a method named [method] with the given [params]. [params] may
 /// contain Futures, in which case this will wait until they've completed before
 /// sending the request.
 ///
 /// The error response is validated using [errorCode] and [errorMessage]. Both
 /// of these must be provided. The error code is checked against [errorCode] and
 /// the error message is checked against [errorMessage]. Either of these may be
 /// matchers.
 ///
 /// Returns a [Future] that completes to the error's [data] field.
 Future expectWebSocketError(String method, Map params, errorCode,
     errorMessage) {
   return schedule(() {
-    return Future.wait([
-      _ensureWebSocket(),
-      awaitObject(params)
-    ]).then((results) {
-      var resolvedParams = results[1];
-      return _jsonRpcRequest(method, resolvedParams);
-    }).then((response) {
+    return webSocketRequest(method, params).then((response) {
       expect(response["error"]["code"], errorCode);
       expect(response["error"]["message"], errorMessage);
-
       return response["error"]["data"];
     });
   }, "send $method with $params to web socket and expect error $errorCode");
 }
 
 /// Validates that [root] was not bound to a port when pub serve started.
 Future expectNotServed(String root) {
   return schedule(() {
     expect(_ports.containsKey(root), isFalse);
   });
 }
 
 /// The next id to use for a JSON-RPC 2.0 request.
 var _rpcId = 0;
 
 /// Sends a JSON-RPC 2.0 request calling [method] with [params].
 ///
 /// Returns the response object.
 Future<Map> _jsonRpcRequest(String method, Map params) {
   var id = _rpcId++;
   _webSocket.add(JSON.encode({
     "jsonrpc": "2.0",
     "method": method,
     "params": params,
     "id": id
   }));
 
-  return _webSocketBroadcastStream.first.then((value) {
-    value = JSON.decode(value);
+  return _webSocketBroadcastStream
+      .firstWhere((response) => response["id"] == id).then((value) {
     currentSchedule.addDebugInfo(
         "Web Socket request $method with params $params\n"
         "Result: $value");
 
     expect(value["id"], equals(id));
     return value;
   });
 }
 
 /// Returns a [Future] that completes to a URL string for the server serving
@@ skipping 19 matching lines @@
 /// included. Unlike [getServerUrl], this should only be called after the ports
 /// are known.
 String _getServerUrlSync([String root, String path]) {
   if (root == null) root = 'web';
   expect(_ports, contains(root));
   var url = "http://127.0.0.1:${_ports[root]}";
   if (path != null) url = "$url/$path";
   return url;
 }