Make StreamSubscription be the active part of a stream.

pkg/scheduled_test/lib/scheduled_process.dart

Index: pkg/scheduled_test/lib/scheduled_process.dart
diff --git a/pkg/scheduled_test/lib/scheduled_process.dart b/pkg/scheduled_test/lib/scheduled_process.dart
index 23370f10942602b5ccc0e42e56dbce3c3d87c70e..071ee384dc8d1a523e21da4b8fdff3022c186015 100644
--- a/pkg/scheduled_test/lib/scheduled_process.dart
+++ b/pkg/scheduled_test/lib/scheduled_process.dart
@@ -39,7 +39,7 @@ class ScheduledProcess {
   Stream<String> _stdoutLog;
 
   /// A line-by-line view of the standard output stream of the process.
-  Stream<String> _stdout;
+  StreamIterator<String> _stdout;
 
   /// A canceller that controls both [_stdout] and [_stdoutLog].
   StreamCanceller _stdoutCanceller;
@@ -49,7 +49,7 @@ class ScheduledProcess {
   Stream<String> _stderrLog;
 
   /// A line-by-line view of the standard error stream of the process.
-  Stream<String> _stderr;
+  StreamIterator<String> _stderr;
 
   /// A canceller that controls both [_stderr] and [_stderrLog].
   StreamCanceller _stderrCanceller;
@@ -102,8 +102,8 @@ class ScheduledProcess {
     _stderrCanceller = stderrWithCanceller.last;
     _stderrLog = stderrWithCanceller.first;
 
-    _stdout = stdoutStream();
-    _stderr = stderrStream();
+    _stdout = new StreamIterator<String>(stdoutStream());
+    _stderr = new StreamIterator<String>(stderrStream());
   }
 
   /// Updates [_description] to reflect [executable] and [arguments], which are
@@ -239,14 +239,47 @@ class ScheduledProcess {
     }, "cleaning up process '$description'");
   }
 
+  /**
+   * Returns the first element of a [StreamIterator].

[floitsch, 2013/05/23 18:38:23]

Remove and use streamIteratorFirst from utils ?

+   *
+   * If the [StreamIterator] has no elements, the result is a state error.
+   */
+  Future<String> _streamIteratorFirst(StreamIterator<String> streamIterator) {
+    StackTrace stackTrace = new Trace().current();
+    return streamIterator.moveNext().then((hasNext) {
+      if (hasNext) {
+        return streamIterator.current;
+      } else {
+        return new Future.error(new StateError("No elements"), stackTrace);
+      }
+    });
+  }
+
   /// Reads the next line of stdout from the process.
-  Future<String> nextLine() => schedule(() => streamFirst(_stdout),
+  Future<String> nextLine() => schedule(() => _streamIteratorFirst(_stdout),
       "reading the next stdout line from process '$description'");
 
   /// Reads the next line of stderr from the process.
-  Future<String> nextErrLine() => schedule(() => streamFirst(_stderr),
+  Future<String> nextErrLine() => schedule(() => _streamIteratorFirst(_stderr),
       "reading the next stderr line from process '$description'");
 
+  /// Collects all remaining lines from a [StreamIterator] of lines.
+  ///
+  /// Returns the concatenation of the collected lines joined by newlines.
+  static Future<String> _concatRest(StreamIterator<String> streamIterator) {

[floitsch, 2013/05/23 18:38:23]

ditto.

+    Future<String> collectAll(List<String> accumulator) {
+      return streamIterator.moveNext().then((hasNext) {
+        if (hasNext) {
+          accumulator.add(streamIterator.current);
+          return collectAll(accumulator);
+        } else {
+          return accumulator.join("\n");
+        }
+      });
+    }
+    return collectAll(<String>[]);
+  }
+
   /// Reads the remaining stdout from the process. This should only be called
   /// after kill() or shouldExit().
   Future<String> remainingStdout() {
@@ -254,8 +287,8 @@ class ScheduledProcess {
       throw new StateError("remainingStdout() should only be called after "
           "kill() or shouldExit().");
     }
-
-    return schedule(() => _stdout.toList().then((lines) => lines.join("\n")),
+    List<String> accumulator = <String>[];

[floitsch, 2013/05/23 18:38:23]

unused.

+    return schedule(() => _concatRest(_stdout),
         "reading the remaining stdout from process '$description'");
   }
 
@@ -267,7 +300,7 @@ class ScheduledProcess {
           "kill() or shouldExit().");
     }
 
-    return schedule(() => _stderr.toList().then((lines) => lines.join("\n")),
+    return schedule(() => _concatRest(_stderr),
         "reading the remaining stderr from process '$description'");
   }