Roll back Pub stream changes.

utils/tests/pub/test_pub.dart

Index: utils/tests/pub/test_pub.dart
diff --git a/utils/tests/pub/test_pub.dart b/utils/tests/pub/test_pub.dart
index 916132af215bedb3465036d89e900bcc21614b29..81bc9e9cceb98505e635080b454dc700668517e9 100644
--- a/utils/tests/pub/test_pub.dart
+++ b/utils/tests/pub/test_pub.dart
@@ -110,7 +110,8 @@ void serve([List<Descriptor> contents]) {
           return;
         }
 
-        stream.toBytes().then((data) {
+        var future = consumeInputStream(stream);
+        future.then((data) {
           response.statusCode = 200;
           response.contentLength = data.length;
           response.outputStream.write(data);
@@ -777,7 +778,7 @@ abstract class Descriptor {
 
   /// Loads the file at [path] from within this descriptor. If [path] is empty,
   /// loads the contents of the descriptor itself.
-  ByteStream load(List<String> path);
+  InputStream load(List<String> path);
 
   /// Schedules the directory to be created before Pub is run with
   /// [schedulePub]. The directory will be created relative to the sandbox
@@ -897,13 +898,16 @@ class FileDescriptor extends Descriptor {
   }
 
   /// Loads the contents of the file.
-  ByteStream load(List<String> path) {
+  InputStream load(List<String> path) {
     if (!path.isEmpty) {
       var joinedPath = Strings.join(path, '/');
       throw "Can't load $joinedPath from within $name: not a directory.";
     }
 
-    return new ByteStream.fromBytes(contents.charCodes);
+    var stream = new ListInputStream();
+    stream.write(contents.charCodes);
+    stream.markEndOfStream();
+    return stream;
   }
 }
 
@@ -955,7 +959,7 @@ class DirectoryDescriptor extends Descriptor {
   }
 
   /// Loads [path] from within this directory.
-  ByteStream load(List<String> path) {
+  InputStream load(List<String> path) {
     if (path.isEmpty) {
       throw "Can't load the contents of $name: is a directory.";
     }
@@ -985,10 +989,10 @@ class FutureDescriptor extends Descriptor {
 
   Future delete(dir) => _future.then((desc) => desc.delete(dir));
 
-  ByteStream load(List<String> path) {
-    var controller = new StreamController<List<int>>();
-    _future.then((desc) => store(desc.load(path), controller));
-    return new ByteStream(controller.stream);
+  InputStream load(List<String> path) {
+    var resultStream = new ListInputStream();
+    _future.then((desc) => pipeInputToInput(desc.load(path), resultStream));
+    return resultStream;
   }
 }
 
@@ -1081,7 +1085,7 @@ class TarFileDescriptor extends Descriptor {
       tempDir = _tempDir;
       return Future.wait(contents.mappedBy((child) => child.create(tempDir)));
     }).then((createdContents) {
-      return createTarGz(createdContents, baseDir: tempDir).toBytes();
+      return consumeInputStream(createTarGz(createdContents, baseDir: tempDir));
     }).then((bytes) {
       return new File(join(parentDir, _stringName)).writeAsBytes(bytes);
     }).then((file) {
@@ -1100,13 +1104,13 @@ class TarFileDescriptor extends Descriptor {
   }
 
   /// Loads the contents of this tar file.
-  ByteStream load(List<String> path) {
+  InputStream load(List<String> path) {
     if (!path.isEmpty) {
       var joinedPath = Strings.join(path, '/');
       throw "Can't load $joinedPath from within $name: not a directory.";
     }
 
-    var controller = new StreamController<List<int>>();
+    var sinkStream = new ListInputStream();
     var tempDir;
     // TODO(rnystrom): Use withTempDir() here.
     // TODO(nweiz): propagate any errors to the return value. See issue 3657.
@@ -1115,11 +1119,11 @@ class TarFileDescriptor extends Descriptor {
       return create(tempDir);
     }).then((tar) {
       var sourceStream = tar.openInputStream();
-      return store(wrapInputStream(sourceStream), controller).then((_) {
+      return pipeInputToInput(sourceStream, sinkStream).then((_) {
         tempDir.delete(recursive: true);
       });
     });
-    return new ByteStream(controller.stream);
+    return sinkStream;
   }
 }
 
@@ -1138,7 +1142,7 @@ class NothingDescriptor extends Descriptor {
     });
   }
 
-  ByteStream load(List<String> path) {
+  InputStream load(List<String> path) {
     if (path.isEmpty) {
       throw "Can't load the contents of $name: it doesn't exist.";
     } else {
@@ -1208,7 +1212,7 @@ class ScheduledProcess {
   final String name;
 
   /// The process future that's scheduled to run.
-  Future<PubProcess> _processFuture;
+  Future<Process> _processFuture;
 
   /// The process that's scheduled to run. It may be null.
   Process _process;
@@ -1216,35 +1220,13 @@ class ScheduledProcess {
   /// The exit code of the scheduled program. It may be null.
   int _exitCode;
 
-  /// A future that will complete to a list of all the lines emitted on the
-  /// process's standard output stream. This is independent of what data is read
-  /// from [_stdout].
-  Future<List<String>> _stdoutLines;
+  /// A [StringInputStream] wrapping the stdout of the process that's scheduled
+  /// to run.
+  final Future<StringInputStream> _stdoutFuture;
 
-  /// A [Stream] of stdout lines emitted by the process that's scheduled to run.
-  /// It may be null.
-  Stream<String> _stdout;
-
-  /// A [Future] that will resolve to [_stdout] once it's available.
-  Future get _stdoutFuture => _processFuture.then((_) => _stdout);
-
-  /// A [StreamSubscription] that controls [_stdout].
-  StreamSubscription _stdoutSubscription;
-
-  /// A future that will complete to a list of all the lines emitted on the
-  /// process's standard error stream. This is independent of what data is read
-  /// from [_stderr].
-  Future<List<String>> _stderrLines;
-
-  /// A [Stream] of stderr lines emitted by the process that's scheduled to run.
-  /// It may be null.
-  Stream<String> _stderr;
-
-  /// A [Future] that will resolve to [_stderr] once it's available.
-  Future get _stderrFuture => _processFuture.then((_) => _stderr);
-
-  /// A [StreamSubscription] that controls [_stderr].
-  StreamSubscription _stderrSubscription;
+  /// A [StringInputStream] wrapping the stderr of the process that's scheduled
+  /// to run.
+  final Future<StringInputStream> _stderrFuture;
 
   /// The exit code of the process that's scheduled to run. This will naturally
   /// only complete once the process has terminated.
@@ -1261,57 +1243,42 @@ class ScheduledProcess {
   bool _endExpected = false;
 
   /// Wraps a [Process] [Future] in a scheduled process.
-  ScheduledProcess(this.name, Future<PubProcess> process)
-    : _processFuture = process {
-    var pairFuture = process.then((p) {
+  ScheduledProcess(this.name, Future<Process> process)
+    : _processFuture = process,
+      _stdoutFuture = process.then((p) => new StringInputStream(p.stdout)),
+      _stderrFuture = process.then((p) => new StringInputStream(p.stderr)) {
+    process.then((p) {
       _process = p;
-
-      var stdoutTee = tee(p.stdout.handleError((e) {
-        registerException(e.error, e.stackTrace);
-      }));
-      var stdoutPair = streamWithSubscription(stdoutTee.last);
-      _stdout = stdoutPair.first;
-      _stdoutSubscription = stdoutPair.last;
-
-      var stderrTee = tee(p.stderr.handleError((e) {
-        registerException(e.error, e.stackTrace);
-      }));
-      var stderrPair = streamWithSubscription(stderrTee.last);
-      _stderr = stderrPair.first;
-      _stderrSubscription = stderrPair.last;
-
-      return new Pair(stdoutTee.first, stderrTee.first);
     });
 
-    _stdoutLines = pairFuture.then((pair) => pair.first.toList());
-    _stderrLines = pairFuture.then((pair) => pair.last.toList());
-
     _schedule((_) {
       if (!_endScheduled) {
         throw new StateError("Scheduled process $name must have shouldExit() "
             "or kill() called before the test is run.");
       }
 
-      return process.then((p) => p.exitCode).then((exitCode) {
-        if (_endExpected) {
-          _exitCode = exitCode;
-          _exitCodeCompleter.complete(exitCode);
-          return;
-        }
-
-        // Sleep for half a second in case _endExpected is set in the next
-        // scheduled event.
-        sleep(500).then((_) {
+      return process.then((p) {
+        p.onExit = (c) {
           if (_endExpected) {
-            _exitCodeCompleter.complete(exitCode);
+            _exitCode = c;
+            _exitCodeCompleter.complete(c);
             return;
           }
 
-          return _printStreams().then((_) {
-            registerException(new ExpectException("Process $name ended "
-                "earlier than scheduled with exit code $exitCode"));
+          // Sleep for half a second in case _endExpected is set in the next
+          // scheduled event.
+          sleep(500).then((_) {
+            if (_endExpected) {
+              _exitCodeCompleter.complete(c);
+              return;
+            }
+
+            _printStreams().then((_) {
+              registerException(new ExpectException("Process $name ended "
+                  "earlier than scheduled with exit code $c"));
+            });
           });
-        });
+        };
       });
     });
 
@@ -1331,15 +1298,15 @@ class ScheduledProcess {
       if (_process == null) return;
       // Ensure that the process is dead and we aren't waiting on any IO.
       _process.kill();
-      _stdoutSubscription.cancel();
-      _stderrSubscription.cancel();
+      _process.stdout.close();
+      _process.stderr.close();
     });
   }
 
   /// Reads the next line of stdout from the process.
   Future<String> nextLine() {
     return _scheduleValue((_) {
-      return timeout(_stdoutFuture.then((stream) => streamFirst(stream)),
+      return timeout(_stdoutFuture.then((stream) => readLine(stream)),
           _SCHEDULE_TIMEOUT,
           "waiting for the next stdout line from process $name");
     });
@@ -1348,7 +1315,7 @@ class ScheduledProcess {
   /// Reads the next line of stderr from the process.
   Future<String> nextErrLine() {
     return _scheduleValue((_) {
-      return timeout(_stderrFuture.then((stream) => streamFirst(stream)),
+      return timeout(_stderrFuture.then((stream) => readLine(stream)),
           _SCHEDULE_TIMEOUT,
           "waiting for the next stderr line from process $name");
     });
@@ -1363,8 +1330,7 @@ class ScheduledProcess {
     }
 
     return _scheduleValue((_) {
-      return timeout(_stdoutFuture.then((stream) => stream.toList())
-              .then((lines) => lines.join("\n")),
+      return timeout(_stdoutFuture.then(consumeStringInputStream),
           _SCHEDULE_TIMEOUT,
           "waiting for the last stdout line from process $name");
     });
@@ -1379,8 +1345,7 @@ class ScheduledProcess {
     }
 
     return _scheduleValue((_) {
-      return timeout(_stderrFuture.then((stream) => stream.toList())
-              .then((lines) => lines.join("\n")),
+      return timeout(_stderrFuture.then(consumeStringInputStream),
           _SCHEDULE_TIMEOUT,
           "waiting for the last stderr line from process $name");
     });
@@ -1389,7 +1354,7 @@ class ScheduledProcess {
   /// Writes [line] to the process as stdin.
   void writeLine(String line) {
     _schedule((_) => _processFuture.then(
-        (p) => p.stdin.write('$line\n'.charCodes)));
+        (p) => p.stdin.writeString('$line\n')));
   }
 
   /// Kills the process, and waits until it's dead.
@@ -1398,7 +1363,7 @@ class ScheduledProcess {
     _schedule((_) {
       _endExpected = true;
       _process.kill();
-      timeout(_exitCodeFuture, _SCHEDULE_TIMEOUT,
+      timeout(_exitCodeCompleter.future, _SCHEDULE_TIMEOUT,
           "waiting for process $name to die");
     });
   }
@@ -1409,7 +1374,7 @@ class ScheduledProcess {
     _endScheduled = true;
     _schedule((_) {
       _endExpected = true;
-      return timeout(_exitCodeFuture, _SCHEDULE_TIMEOUT,
+      return timeout(_exitCodeCompleter.future, _SCHEDULE_TIMEOUT,
           "waiting for process $name to exit").then((exitCode) {
         if (expectedExitCode != null) {
           expect(exitCode, equals(expectedExitCode));
@@ -1419,21 +1384,24 @@ class ScheduledProcess {
   }
 
   /// Prints the remaining data in the process's stdout and stderr streams.
-  /// Prints nothing if the streams are empty.
+  /// Prints nothing if the straems are empty.
   Future _printStreams() {
-    void printStream(String streamName, List<String> lines) {
-      if (lines.isEmpty) return;
+    Future printStream(String streamName, StringInputStream stream) {
+      return consumeStringInputStream(stream).then((output) {
+        if (output.isEmpty) return;
 
-      print('\nProcess $name $streamName:');
-      for (var line in lines) {
-        print('| $line');
-      }
+        print('\nProcess $name $streamName:');
+        for (var line in output.trim().split("\n")) {
+          print('| $line');
+        }
+        return;
+      });
     }
 
-    return _stdoutLines.then((stdoutLines) {
-      printStream('stdout', stdoutLines);
-      return _stderrLines.then((stderrLines) {
-        printStream('stderr', stderrLines);
+    return _stdoutFuture.then((stdout) {
+      return _stderrFuture.then((stderr) {
+        return printStream('stdout', stdout)
+            .then((_) => printStream('stderr', stderr));
       });
     });
   }