Use the dart:async Stream API thoroughly in Pub.

utils/tests/pub/test_pub.dart

 // Copyright (c) 2012, the Dart project authors.  Please see the AUTHORS 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.
 
 /// Test infrastructure for testing pub. Unlike typical unit tests, most pub
 /// tests are integration tests that stage some stuff on the file system, run
 /// pub, and then validate the results. This library provides an API to build
 /// tests like that.
 library test_pub;
 
@@ skipping 77 matching lines @@
         var stream;
         try {
           stream = baseDir.load(path);
         } catch (e) {
           response.statusCode = 404;
           response.contentLength = 0;
           response.outputStream.close();
           return;
         }
 
-        var future = consumeInputStream(stream);
-        future.then((data) {
+        stream.toBytes().then((data) {
           response.statusCode = 200;
           response.contentLength = data.length;
           response.outputStream.write(data);
           response.outputStream.close();
         }).catchError((e) {
           print("Exception while handling ${request.uri}: $e");
           response.statusCode = 500;
           response.reasonPhrase = e.message;
           response.outputStream.close();
         });
@@ skipping 648 matching lines @@
   /// system entry within [dir]. Returns a [Future] that completes to `null` if
   /// the entry is valid, or throws an error if it failed.
   Future validate(String dir);
 
   /// Deletes the file or directory within [dir]. Returns a [Future] that is
   /// completed after the deletion is done.
   Future delete(String dir);
 
   /// Loads the file at [path] from within this descriptor. If [path] is empty,
   /// loads the contents of the descriptor itself.
-  InputStream load(List<String> path);
+  ByteStream load(List<String> path);
 
   /// Schedules the directory to be created before Pub is run with [runPub].
   /// The directory will be created relative to the sandbox directory.
   // TODO(nweiz): Use implicit closurization once issue 2984 is fixed.
   void scheduleCreate() => _schedule((dir) => this.create(dir));
 
   /// Schedules the file or directory to be deleted recursively.
   void scheduleDelete() => _schedule((dir) => this.delete(dir));
 
   /// Schedules the directory to be validated after Pub is run with [runPub].
@@ skipping 94 matching lines @@
       return readTextFile(file).then((text) {
         if (text == contents) return null;
 
         Expect.fail('File $file should contain:\n\n$contents\n\n'
                     'but contained:\n\n$text');
       });
     });
   }
 
   /// Loads the contents of the file.
-  InputStream load(List<String> path) {
+  ByteStream load(List<String> path) {
     if (!path.isEmpty) {
       var joinedPath = Strings.join(path, '/');
       throw "Can't load $joinedPath from within $name: not a directory.";
     }
 
-    var stream = new ListInputStream();
-    stream.write(contents.charCodes);
-    stream.markEndOfStream();
-    return stream;
+    return new ByteStream.fromBytes(contents.charCodes);
   }
 }
 
 /// Describes a directory and its contents. These are used both for setting up
 /// an expected directory tree before running a test, and for validating that
 /// the file system matches some expectations after running it.
 class DirectoryDescriptor extends Descriptor {
   /// The files and directories contained in this directory.
   final List<Descriptor> contents;
 
@@ skipping 31 matching lines @@
       // Validate each of the items in this directory.
       final entryFutures =
           contents.mappedBy((entry) => entry.validate(dir)).toList();
 
       // If they are all valid, the directory is valid.
       return Future.wait(entryFutures).then((entries) => null);
     });
   }
 
   /// Loads [path] from within this directory.
-  InputStream load(List<String> path) {
+  ByteStream load(List<String> path) {
     if (path.isEmpty) {
       throw "Can't load the contents of $name: is a directory.";
     }
 
     for (var descriptor in contents) {
       if (descriptor.name == path[0]) {
         return descriptor.load(path.getRange(1, path.length - 1));
       }
     }
 
     throw "Directory $name doesn't contain ${Strings.join(path, '/')}.";
   }
 }
 
 /// Wraps a [Future] that will complete to a [Descriptor] and makes it behave
 /// like a concrete [Descriptor]. This is necessary when the contents of the
 /// descriptor depends on information that's not available until part of the
 /// test run is completed.
 class FutureDescriptor extends Descriptor {
   Future<Descriptor> _future;
 
   FutureDescriptor(this._future) : super('<unknown>');
 
   Future create(dir) => _future.then((desc) => desc.create(dir));
 
   Future validate(dir) => _future.then((desc) => desc.validate(dir));
 
   Future delete(dir) => _future.then((desc) => desc.delete(dir));
 
-  InputStream load(List<String> path) {
-    var resultStream = new ListInputStream();
-    _future.then((desc) => pipeInputToInput(desc.load(path), resultStream));
-    return resultStream;
+  ByteStream load(List<String> path) {
+    var controller = new StreamController<List<int>>();
+    _future.then((desc) => store(desc.load(path), controller));
+    return new ByteStream(controller.stream);
   }
 }
 
 /// Describes a Git repository and its contents.
 class GitRepoDescriptor extends DirectoryDescriptor {
   GitRepoDescriptor(Pattern name, List<Descriptor> contents)
   : super(name, contents);
 
   /// Creates the Git repository and commits the contents.
   Future<Directory> create(parentDir) {
@@ skipping 79 matching lines @@
 
   /// Creates the files and directories within this tar file, then archives
   /// them, compresses them, and saves the result to [parentDir].
   Future<File> create(parentDir) {
     // TODO(rnystrom): Use withTempDir().
     var tempDir;
     return createTempDir().then((_tempDir) {
       tempDir = _tempDir;
       return Future.wait(contents.mappedBy((child) => child.create(tempDir)));
     }).then((createdContents) {
-      return consumeInputStream(createTarGz(createdContents, baseDir: tempDir));
+      return createTarGz(createdContents, baseDir: tempDir).toBytes();
     }).then((bytes) {
       return new File(join(parentDir, _stringName)).writeAsBytes(bytes);
     }).then((file) {
       return deleteDir(tempDir).then((_) => file);
     });
   }
 
   /// Validates that the `.tar.gz` file at [path] contains the expected
   /// contents.
   Future validate(String path) {
     throw "TODO(nweiz): implement this";
   }
 
   Future delete(dir) {
     throw new UnsupportedError('');
   }
 
   /// Loads the contents of this tar file.
-  InputStream load(List<String> path) {
+  ByteStream load(List<String> path) {
     if (!path.isEmpty) {
       var joinedPath = Strings.join(path, '/');
       throw "Can't load $joinedPath from within $name: not a directory.";
     }
 
-    var sinkStream = new ListInputStream();
+    var controller = new StreamController<List<int>>();
     var tempDir;
     // TODO(rnystrom): Use withTempDir() here.
     // TODO(nweiz): propagate any errors to the return value. See issue 3657.
     createTempDir().then((_tempDir) {
       tempDir = _tempDir;
       return create(tempDir);
     }).then((tar) {
       var sourceStream = tar.openInputStream();
-      return pipeInputToInput(sourceStream, sinkStream).then((_) {
+      return store(wrapInputStream(sourceStream), controller).then((_) {
         tempDir.delete(recursive: true);
       });
     });
-    return sinkStream;
+    return new ByteStream(controller.stream);
   }
 }
 
 /// A descriptor that validates that no file exists with the given name.
 class NothingDescriptor extends Descriptor {
   NothingDescriptor(String name) : super(name);
 
   Future create(dir) => new Future.immediate(null);
   Future delete(dir) => new Future.immediate(null);
 
   Future validate(String dir) {
     return exists(join(dir, name)).then((exists) {
       if (exists) Expect.fail('File $name in $dir should not exist.');
     });
   }
 
-  InputStream load(List<String> path) {
+  ByteStream load(List<String> path) {
     if (path.isEmpty) {
       throw "Can't load the contents of $name: it doesn't exist.";
     } else {
       throw "Can't load ${Strings.join(path, '/')} from within $name: $name "
         "doesn't exist.";
     }
   }
 }
 
 /// A function that creates a [Validator] subclass.
@@ skipping 50 matching lines @@
 /// Before running the test, either [shouldExit] or [kill] must be called on
 /// this to ensure that the process terminates when expected.
 ///
 /// If the test fails, this will automatically print out any remaining stdout
 /// and stderr from the process to aid debugging.
 class ScheduledProcess {
   /// The name of the process. Used for error reporting.
   final String name;
 
   /// The process future that's scheduled to run.
-  Future<Process> _processFuture;
+  Future<PubProcess> _processFuture;
 
   /// The process that's scheduled to run. It may be null.
   Process _process;
 
   /// The exit code of the scheduled program. It may be null.
   int _exitCode;
 
-  /// A [StringInputStream] wrapping the stdout of the process that's scheduled
-  /// to run.
-  final Future<StringInputStream> _stdoutFuture;
+  /// 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 stderr of the process that's scheduled
-  /// to run.
-  final Future<StringInputStream> _stderrFuture;
+  /// 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;
 
   /// The exit code of the process that's scheduled to run. This will naturally
   /// only complete once the process has terminated.
   Future<int> get _exitCodeFuture => _exitCodeCompleter.future;
 
   /// The completer for [_exitCode].
   final Completer<int> _exitCodeCompleter = new Completer();
 
   /// Whether the user has scheduled the end of this process by calling either
   /// [shouldExit] or [kill].
   bool _endScheduled = false;
 
   /// Whether the process is expected to terminate at this point.
   bool _endExpected = false;
 
   /// Wraps a [Process] [Future] in a scheduled process.
-  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) {
+  ScheduledProcess(this.name, Future<PubProcess> process)
+    : _processFuture = process {
+    var pairFuture = 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.onExit = (c) {
+      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((_) {
           if (_endExpected) {
-            _exitCode = c;
-            _exitCodeCompleter.complete(c);
+            _exitCodeCompleter.complete(exitCode);
             return;
           }
 
-          // 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"));
-            });
+          return _printStreams().then((_) {
+            registerException(new ExpectException("Process $name ended "
+                "earlier than scheduled with exit code $exitCode"));
           });
-        };
+        });
       });
     });
 
     _scheduleOnException((_) {
       if (_process == null) return;
 
       if (_exitCode == null) {
         print("\nKilling process $name prematurely.");
         _endExpected = true;
         _process.kill();
       }
 
       return _printStreams();
     });
 
     _scheduleCleanup((_) {
       if (_process == null) return;
       // Ensure that the process is dead and we aren't waiting on any IO.
       _process.kill();
-      _process.stdout.close();
-      _process.stderr.close();
+      _stdoutSubscription.cancel();
+      _stderrSubscription.cancel();
     });
   }
 
   /// Reads the next line of stdout from the process.
   Future<String> nextLine() {
     return _scheduleValue((_) {
-      return timeout(_stdoutFuture.then((stream) => readLine(stream)),
+      return timeout(_stdoutFuture.then((stream) => streamFirst(stream)),
           _SCHEDULE_TIMEOUT,
           "waiting for the next stdout line from process $name");
     });
   }
 
   /// Reads the next line of stderr from the process.
   Future<String> nextErrLine() {
     return _scheduleValue((_) {
-      return timeout(_stderrFuture.then((stream) => readLine(stream)),
+      return timeout(_stderrFuture.then((stream) => streamFirst(stream)),
           _SCHEDULE_TIMEOUT,
           "waiting for the next stderr line from process $name");
     });
   }
 
   /// Reads the remaining stdout from the process. This should only be called
   /// after kill() or shouldExit().
   Future<String> remainingStdout() {
     if (!_endScheduled) {
       throw new StateError("remainingStdout() should only be called after "
           "kill() or shouldExit().");
     }
 
     return _scheduleValue((_) {
-      return timeout(_stdoutFuture.then(consumeStringInputStream),
+      return timeout(_stdoutFuture.then((stream) => stream.toList())
+              .then((lines) => lines.join("\n")),
           _SCHEDULE_TIMEOUT,
           "waiting for the last stdout line from process $name");
     });
   }
 
   /// Reads the remaining stderr from the process. This should only be called
   /// after kill() or shouldExit().
   Future<String> remainingStderr() {
     if (!_endScheduled) {
       throw new StateError("remainingStderr() should only be called after "
           "kill() or shouldExit().");
     }
 
     return _scheduleValue((_) {
-      return timeout(_stderrFuture.then(consumeStringInputStream),
+      return timeout(_stderrFuture.then((stream) => stream.toList())
+              .then((lines) => lines.join("\n")),
           _SCHEDULE_TIMEOUT,
           "waiting for the last stderr line from process $name");
     });
   }
 
   /// Writes [line] to the process as stdin.
   void writeLine(String line) {
     _schedule((_) => _processFuture.then(
-        (p) => p.stdin.writeString('$line\n')));
+        (p) => p.stdin.write('$line\n'.charCodes)));
   }
 
   /// Kills the process, and waits until it's dead.
   void kill() {
     _endScheduled = true;
     _schedule((_) {
       _endExpected = true;
       _process.kill();
-      timeout(_exitCodeCompleter.future, _SCHEDULE_TIMEOUT,
+      timeout(_exitCodeFuture, _SCHEDULE_TIMEOUT,
           "waiting for process $name to die");
     });
   }
 
   /// Waits for the process to exit, and verifies that the exit code matches
   /// [expectedExitCode] (if given).
   void shouldExit([int expectedExitCode]) {
     _endScheduled = true;
     _schedule((_) {
       _endExpected = true;
-      return timeout(_exitCodeCompleter.future, _SCHEDULE_TIMEOUT,
+      return timeout(_exitCodeFuture, _SCHEDULE_TIMEOUT,
           "waiting for process $name to exit").then((exitCode) {
         if (expectedExitCode != null) {
           expect(exitCode, equals(expectedExitCode));
         }
       });
     });
   }
 
   /// Prints the remaining data in the process's stdout and stderr streams.
-  /// Prints nothing if the straems are empty.
+  /// Prints nothing if the streams are empty.
   Future _printStreams() {
-    Future printStream(String streamName, StringInputStream stream) {
-      return consumeStringInputStream(stream).then((output) {
-        if (output.isEmpty) return;
+    void printStream(String streamName, List<String> lines) {
+      if (lines.isEmpty) return;
 
-        print('\nProcess $name $streamName:');
-        for (var line in output.trim().split("\n")) {
-          print('| $line');
-        }
-        return;
-      });
+      print('\nProcess $name $streamName:');
+      for (var line in lines) {
+        print('| $line');
+      }
     }
 
-    return _stdoutFuture.then((stdout) {
-      return _stderrFuture.then((stderr) {
-        return printStream('stdout', stdout)
-            .then((_) => printStream('stderr', stderr));
+    return _stdoutLines.then((stdoutLines) {
+      printStream('stdout', stdoutLines);
+      return _stderrLines.then((stderrLines) {
+        printStream('stderr', stderrLines);
       });
     });
   }
 }
 
 /// A class representing an [HttpServer] that's scheduled to run in the course
 /// of the test. This class allows the server's request handling to be
 /// scheduled synchronously. All operations on this class are scheduled.
 class ScheduledServer {
   /// The wrapped server.
@@ skipping 136 matching lines @@
 /// calling [completion] is unnecessary.
 void expectLater(Future actual, matcher, {String reason,
     FailureHandler failureHandler, bool verbose: false}) {
   _schedule((_) {
     return actual.then((value) {
       expect(value, matcher, reason: reason, failureHandler: failureHandler,
           verbose: false);
     });
   });
 }