Change ScheduledProcess's output streams over to be ScheduledStreams.

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';
 
 import 'package:http/http.dart' as http;
 import 'package:path/path.dart' as p;
 import 'package:scheduled_test/scheduled_process.dart';
+import 'package:scheduled_test/scheduled_stream.dart';
 import 'package:scheduled_test/scheduled_test.dart';
 
 import '../test_pub.dart';
 
 /// The pub process running "pub serve".
 ScheduledProcess _pubServer;
 
 /// The ephemeral port assigned to the running server.
 int _port;
 
@@ skipping 84 matching lines @@
   // Dart2js can take a long time to compile dart code, so we increase the
   // timeout to cope with that.
   currentSchedule.timeout *= 1.5;
 
   return startPub(args: pubArgs);
 }
 
 /// Schedules starting the "pub serve" process and records its port number for
 /// future requests.
 ///
-/// If [shouldGetFirst] is `true`, validates that pub get is run first. In that
-/// case, you can also pass [numDownloads] to specify how many packages should
-/// be downloaded during the get.
+/// If [shouldGetFirst] is `true`, validates that pub get is run first.
 ///
 /// Returns the `pub serve` process.
-ScheduledProcess pubServe({bool shouldGetFirst: false,
-    Iterable<String> args, int numDownloads: 0}) {
+ScheduledProcess pubServe({bool shouldGetFirst: false, Iterable<String> args}) {
   _pubServer = startPubServe(args);
 
   currentSchedule.onComplete.schedule(() {
     if (_webSocket != null) {
       _webSocket.close();
       _webSocket = null;
       _webSocketBroadcastStream = null;
     }
   });
 
   if (shouldGetFirst) {
-    expect(_pubServer.nextLine(),
-        completion(anyOf(
-             startsWith("Your pubspec has changed"),
-             startsWith("You don't have a lockfile"),
-             startsWith("You are missing some dependencies"))));
-    expect(_pubServer.nextLine(),
-        completion(startsWith("Resolving dependencies...")));
-
-    for (var i = 0; i < numDownloads; i++) {
-      expect(_pubServer.nextLine(),
-          completion(startsWith("Downloading")));
-    }
-
-    expect(_pubServer.nextLine(),
-        completion(equals("Got dependencies!")));
+    _pubServer.stdout.expect(consumeThrough("Got dependencies!"));
   }
 
-  expect(_pubServer.nextLine().then(_parsePort), completes);
+  expect(schedule(() => _pubServer.stdout.next()).then(_parsePort), completes);
   return _pubServer;
 }
 
 /// Parses the port number from the "Serving blah on 127.0.0.1:1234" line
 /// printed by pub serve.
 void _parsePort(String line) {
   var match = new RegExp(r"127\.0\.0\.1:(\d+)").firstMatch(line);
   assert(match != null);
   _port = int.parse(match[1]);
 }
@@ skipping 52 matching lines @@
       expect(response.statusCode, equals(405));
     });
   }, "request $urlPath");
 }
 
 /// Reads lines from pub serve's stdout until it prints the build success
 /// message.
 ///
 /// The schedule will not proceed until the output is found. If not found, it
 /// will eventually time out.
-void waitForBuildSuccess() {
-  nextLine() {
-    return _pubServer.nextLine().then((line) {
-      if (line.contains("successfully")) return null;
-
-      // This line wasn't it, so ignore it and keep trying.
-      return nextLine();
-    });
-  }
-
-  schedule(nextLine);
-}
+void waitForBuildSuccess() =>
+  _pubServer.stdout.expect(consumeThrough(contains("successfully")));
 
 /// Schedules opening a web socket connection to the currently running pub
 /// serve.
 Future _ensureWebSocket() {
   // Use the existing one if already connected.
   if (_webSocket != null) return new Future.value();
 
   // Server should already be running.
   assert(_pubServer != null);
   assert(_port != null);
@@ skipping 13 matching lines @@
 /// socket. It omitted, request is JSON encoded to a string first.
 void webSocketShouldReply(request, expectation, {bool encodeRequest: true}) {
   schedule(() => _ensureWebSocket().then((_) {
     if (encodeRequest) request = JSON.encode(request);
     _webSocket.add(request);
     return _webSocketBroadcastStream.first.then((value) {
       expect(JSON.decode(value), expectation);
     });
   }), "send $request to web socket and expect reply that $expectation");
 }