Make StreamSubscription be the active part of a stream.

pkg/scheduled_test/lib/scheduled_process.dart

 // Copyright (c) 2013, 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.
 
 library scheduled_test.scheduled_process;
 
 import 'dart:async';
 import 'dart:io';
 
 import 'scheduled_test.dart';
@@ skipping 21 matching lines @@
   final Encoding _encoding;
 
   /// The process that's scheduled to run.
   ValueFuture<Process> _process;
 
   /// A fork of [_stdout] that records the standard output of the process. Used
   /// for debugging information.
   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;
 
   /// A fork of [_stderr] that records the standard error of the process. Used
   /// for debugging information.
   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;
 
   /// The exit code of the process that's scheduled to run. This will naturally
   /// only complete once the process has terminated.
   ValueFuture<int> _exitCode;
 
   /// Whether the user has scheduled the end of this process by calling either
   /// [shouldExit] or [kill].
@@ skipping 32 matching lines @@
     var stdoutWithCanceller = _lineStreamWithCanceller(
         _process.then((p) => p.stdout));
     _stdoutCanceller = stdoutWithCanceller.last;
     _stdoutLog = stdoutWithCanceller.first;
 
     var stderrWithCanceller = _lineStreamWithCanceller(
         _process.then((p) => p.stderr));
     _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
   /// the same values as in [start].
   void _updateDescription(executable, arguments) {
     if (_explicitDescription) return;
     if (executable is Future) {
       _description = "future process";
     } else if (arguments is Future || arguments.any((e) => e is Future)) {
       _description = executable;
@@ skipping 116 matching lines @@
             "$exitDescription\n"
             "Standard output:\n"
             "${prefixLines(stdout)}\n"
             "Standard error:\n"
             "${prefixLines(stderr)}");
       });
     }, "cleaning up process '$description'");
   }
 
   /// 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'");
 
   /// 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 schedule(() => _stdout.toList().then((lines) => lines.join("\n")),
+    return schedule(() => concatRest(_stdout),
         "reading the remaining stdout from process '$description'");
   }
 
   /// 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 schedule(() => _stderr.toList().then((lines) => lines.join("\n")),
+    return schedule(() => concatRest(_stderr),
         "reading the remaining stderr from process '$description'");
   }
 
   /// Returns a stream that will emit anything the process emits via the
   /// process's standard output from now on.
   ///
   /// This stream will be independent from any other methods that deal with
   /// standard output, including other calls to [stdoutStream].
   ///
   /// This can be overridden by subclasses to return a derived standard output
@@ skipping 55 matching lines @@
     schedule(() {
       _endExpected = true;
       return _exitCode.then((exitCode) {
         if (expectedExitCode != null) {
           expect(exitCode, equals(expectedExitCode));
         }
       });
     }, "waiting for process '$description' to exit");
   }
 }