Fix ALL the pub tests.

utils/pub/io.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.
 
 /// Helper functionality to make working with IO easier.
 library io;
 
 import 'dart:async';
 import 'dart:io';
 import 'dart:isolate';
@@ skipping 170 matching lines @@
   log.fine("Ensuring directory $path exists.");
   if (path == '.') return new Future.immediate(new Directory('.'));
 
   return dirExists(path).then((exists) {
     if (exists) {
       log.fine("Directory $path already exists.");
       return new Future.immediate(new Directory(path));
     }
 
     return ensureDir(dirname(path)).then((_) {
-      return createDir(path)
-        .catchError((error) {
-          if (error is! DirectoryIOException) return false;
-          // Error 17 means the directory already exists (or 183 on Windows).
-          if (error.osError.errorCode != 17 &&
-              error.osError.errorCode != 183) {
+      return createDir(path).catchError((asyncError) {
+        var error = getRealError(asyncError);
+        if (error is! DirectoryIOException) throw asyncError;
+        // Error 17 means the directory already exists (or 183 on Windows).
+        if (error.osError.errorCode != 17 &&
+            error.osError.errorCode != 183) {
           log.fine("Got 'already exists' error when creating directory.");

[Bob Nystrom, 2013/01/09 16:49:23]

This log is wrong. This "if" is for when the error is *not* an "already exists"
one. We should probably negate the condition:

if (error.osError.errorCode == 17 ||
    error.osError.errorCode == 183) {
  log.fine("Got 'already exists' error when creating directory.");
  return _getDirectory(path);
}

throw asyncError;

-          return false;
+          throw asyncError;
         }
 
-          return _getDirectory(path);
-        });
+        return _getDirectory(path);
+      });
     });
   });
 }
 
 /// Creates a temp directory whose name will be based on [dir] with a unique
 /// suffix appended to it. If [dir] is not provided, a temp directory will be
 /// created in a platform-dependent temporary location. Returns a [Future] that
 /// completes when the directory is created.
 Future<Directory> createTempDir([dir = '']) {
   dir = _getDirectory(dir);
@@ skipping 342 matching lines @@
   }
 
   var completer = new Completer<String>();
   var buffer = new StringBuffer();
   stream.onClosed = () => completer.complete(buffer.toString());
   stream.onData = () => buffer.add(stream.read());
   stream.onError = (e) => completer.completeError(e, stackTrace);
   return completer.future;
 }
 
-/// Wrap an InputStream in a ListInputStream. This eagerly drains the [source]
-/// input stream. This is useful for spawned processes which will not exit until
-/// their output streams have been drained.
-/// TODO(rnystrom): We should use this logic anywhere we spawn a process.
-InputStream wrapInputStream(InputStream source) {
-  var sink = new ListInputStream();
-  pipeInputToInput(source, sink);
-  return sink;
+/// Wraps [stream] in a single-subscription [ByteStream] that emits the same
+/// data.
+Stream<List<int>> wrapInputStream(InputStream stream) {

[Bob Nystrom, 2013/01/09 16:49:23]

Type the return as a ByteStream?

Also, can we have a more explicit name for this? At the call sites, it wasn't
clear to me what it was wrapping the input stream in. Is this the same as
"toByteStream" in HTTP? Use the same name?

+  if (stream.closed) return new StreamController.singleSubscription()..close();

[Bob Nystrom, 2013/01/09 16:49:23]

The duplicate constructor is a bit gross. How about just:

var controller = new StreamController.singleSubscription();

if (stream.closed) return controller..close();

stream.onClosed = ...
...

+
+  var controller = new StreamController.singleSubscription();
+  stream.onClosed = controller.close;
+  stream.onData = () => controller.add(stream.read());
+  stream.onError = (e) => controller.signalError(new AsyncError(e));
+  return controller.stream;
+}
+
+// TODO(nweiz): remove this ASAP

[Bob Nystrom, 2013/01/09 16:49:23]

Tracking bug?

+/// Wraps [stream] in an [InputStream].
+InputStream wrapByteStream(Stream<List<int>> stream) {

[Bob Nystrom, 2013/01/09 16:49:23]

Maybe these method names should be more explicit. How about "toInputStream".

+  var inputStream = new ListInputStream();
+  stream.listen((chunk) => inputStream.write(chunk),
+      onDone: inputStream.markEndOfStream);
+  return inputStream;
+}
+
+/// Wraps [stream] in a [StreamConsumer] so that [Stream]s can by piped into it
+/// using [Stream.pipe].
+StreamConsumer<List<int>, dynamic> wrapOutputStream(OutputStream stream) =>

[Bob Nystrom, 2013/01/09 16:49:23]

"toStreamConsumer" ?

+  new _OutputStreamConsumer(stream);
+
+/// A [StreamConsumer] that pipes data into an [OutputStream].
+class _OutputStreamConsumer implements StreamConsumer<List<int>, dynamic> {
+  final OutputStream _outputStream;
+
+  _OutputStreamConsumer(this._outputStream)
+    : super();

[Bob Nystrom, 2013/01/09 16:49:23]

Unnecessary, especially since you aren't extending StreamConsumer. Even if you
were, I think it will do this implicitly if you omit it.

+
+  Future consume(Stream<List<int>> stream) {
+    // TODO(nweiz): we have to manually keep track of whether or not the
+    // completer has completed since the output stream could signal an error
+    // after close() has been called but before it has shut down internally. See
+    // the following TODO.
+    var completed = false;
+    var completer = new Completer();
+    stream.listen((data) => _outputStream.write(data), onDone: () {
+      _outputStream.close();
+      // TODO(nweiz): wait until _outputStream.onClosed is called once issue
+      // 7761 is fixed.
+      if (!completed) completer.complete(null);
+      completed = true;
+    });
+
+    _outputStream.onError = (e) {
+      if (!completed) completer.completeError(e);

[Bob Nystrom, 2013/01/09 16:49:23]

Want to do that little hack to grab a stack trace for this?

+      completed = true;
+    };
+
+    return completer.future;
+  }
 }
 
 /// Spawns and runs the process located at [executable], passing in [args].
 /// Returns a [Future] that will complete with the results of the process after
 /// it has ended.
 ///
 /// The spawned process will inherit its parent's environment variables. If
 /// [environment] is provided, that will be used to augment (not replace) the
 /// the inherited variables.
 Future<PubProcessResult> runProcess(String executable, List<String> args,
@@ skipping 35 matching lines @@
   final InputStream stdout;
 
   OutputStream get stdin => _process.stdin;
 
   void set onExit(void callback(int exitCode)) {
     _process.onExit = callback;
   }
 
   _WrappedProcess(Process process)
     : _process = process,
-      stderr = wrapInputStream(process.stderr),
-      stdout = wrapInputStream(process.stdout);
+      stderr = _wrapInputStream(process.stderr),
+      stdout = _wrapInputStream(process.stdout);
 
   bool kill([ProcessSignal signal = ProcessSignal.SIGTERM]) =>
     _process.kill(signal);
+
+  /// Wrap an InputStream in a ListInputStream. This eagerly drains the [source]
+  /// input stream. This is useful for spawned processes which will not exit until

[Bob Nystrom, 2013/01/09 16:49:23]

Long line.

+  /// their output streams have been drained.
+  /// TODO(rnystrom): We should use this logic anywhere we spawn a process.
+  static InputStream _wrapInputStream(InputStream source) {
+    var sink = new ListInputStream();
+    pipeInputToInput(source, sink);
+    return sink;
+  }
 }
 
 /// Calls [fn] with appropriately modified arguments. [fn] should have the same
 /// signature as [Process.start], except that the returned [Future] may have a
 /// type other than [Process].
 Future _doProcess(Function fn, String executable, List<String> args, workingDir,
     Map<String, String> environment) {
   // TODO(rnystrom): Should dart:io just handle this?
   // Spawning a process on Windows will not look for the executable in the
   // system path. So, if executable looks like it needs that (i.e. it doesn't
@@ skipping 28 matching lines @@
 /// Note that timing out will not cancel the asynchronous operation behind
 /// [input].
 Future timeout(Future input, int milliseconds, String description) {
   bool completed = false;
   var completer = new Completer();
   var timer = new Timer(milliseconds, (_) {
     completed = true;
     completer.completeError(new TimeoutException(
         'Timed out while $description.'));
   });
-  input
-    .then((value) {
-      if (completed) return;
-      timer.cancel();
-      completer.complete(value);
-    })
-    .catchError((e) {
-      if (completed) return;
-      timer.cancel();
-      completer.completeError(e.error, e.stackTrace);
-    });
+  input.then((value) {
+    if (completed) return;
+    timer.cancel();
+    completer.complete(value);
+  }).catchError((e) {
+    if (completed) return;
+    timer.cancel();
+    completer.completeError(e.error, e.stackTrace);
+  });
   return completer.future;
 }
 
 /// Creates a temporary directory and passes its path to [fn]. Once the [Future]
 /// returned by [fn] completes, the temporary directory and all its contents
 /// will be deleted.
 Future withTempDir(Future fn(String path)) {
   var tempDir;
-  var future = createTempDir().then((dir) {
+  return asyncWhenComplete(createTempDir().then((dir) {
     tempDir = dir;
     return fn(tempDir.path);
+  }), () {
+    log.fine('Cleaning up temp directory ${tempDir.path}.');
+    return deleteDir(tempDir);
   });
-  future.catchError((_) {}).then((_) {
-    log.fine('Cleaning up temp directory ${tempDir.path}.');
-    deleteDir(tempDir);
-  });
-  return future;
 }
 
 /// Tests whether or not the git command-line app is available for use.
 Future<bool> get isGitInstalled {
   if (_isGitInstalledCache != null) {
     // TODO(rnystrom): The sleep is to pump the message queue. Can use
     // Future.immediate() when #3356 is fixed.
     return sleep(0).then((_) => _isGitInstalledCache);
   }
 
@@ skipping 248 matching lines @@
 Directory _getDirectory(entry) {
   if (entry is Directory) return entry;
   return new Directory(entry);
 }
 
 /// Gets a [Uri] for [uri], which can either already be one, or be a [String].
 Uri _getUri(uri) {
   if (uri is Uri) return uri;
   return new Uri.fromString(uri);
 }