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 102 matching lines @@
 
   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...")));
-
+    _pubServer.stdout.expect(anyOf(
+         startsWith("Your pubspec has changed"),
+         startsWith("You don't have a lockfile"),
+         startsWith("You are missing some dependencies")));
+    _pubServer.stdout.expect(startsWith("Resolving dependencies..."));
     for (var i = 0; i < numDownloads; i++) {
-      expect(_pubServer.nextLine(),
-          completion(startsWith("Downloading")));
+      _pubServer.stdout.expect(startsWith("Downloading"));

[Bob Nystrom, 2014/02/14 17:55:02]

I hacked in the numDownloads stuff because at the time there was no easy way to
skip an unknown number of lines. You can probably delete all of this and just
do:

_pubServer.stdout.expect(consumeThrough("Got dependencies!"));

[nweiz, 2014/02/18 22:01:10]

Done.

     }
 
-    expect(_pubServer.nextLine(),
-        completion(equals("Got dependencies!")));
+    _pubServer.stdout.expect(equals("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")));

[Bob Nystrom, 2014/02/14 17:55:02]

\o/

 
 /// 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");
 }