Change ScheduledProcess's output streams over to be ScheduledStreams.

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:convert';
 import 'dart:io';
 
 import 'package:stack_trace/stack_trace.dart';
 
+import 'scheduled_stream.dart';
 import 'scheduled_test.dart';
 import 'src/utils.dart';
 import 'src/value_future.dart';
 
 /// A class representing a [Process] that is scheduled to run in the course of
 /// the test. This class allows actions on the process to be scheduled
 /// synchronously. All operations on this class are scheduled.
 ///
 /// Before running the test, either [shouldExit] or [kill] must be called on
 /// this to ensure that the process terminates when expected. Note that [kill]
@@ skipping 14 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.
-  StreamIterator<String> _stdout;
+  ScheduledStream<String> get stdout => _stdout;
+  ScheduledStream<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.
-  StreamIterator<String> _stderr;
+  ScheduledStream<String> get stderr => _stderr;
+  ScheduledStream<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 37 matching lines @@
     var stdoutWithCanceller = _lineStreamWithCanceller(
         _process.then((p) => Chain.track(p.stdout)));
     _stdoutCanceller = stdoutWithCanceller.last;
     _stdoutLog = stdoutWithCanceller.first;
 
     var stderrWithCanceller = _lineStreamWithCanceller(
         _process.then((p) => Chain.track(p.stderr)));
     _stderrCanceller = stderrWithCanceller.last;
     _stderrLog = stderrWithCanceller.first;
 
-    _stdout = new StreamIterator<String>(stdoutStream());
-    _stderr = new StreamIterator<String>(stderrStream());
+    _stdout = new ScheduledStream<String>(stdoutStream());
+    _stderr = new ScheduledStream<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 117 matching lines @@
             "Results of running '$description':\n"
             "$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(() => streamIteratorFirst(_stdout),
-      "reading the next stdout line from process '$description'");
-
-  /// Reads the next line of stderr from the process.
-  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(() => 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(() => 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
-  /// stream. This stream will then be used for [nextLine], [nextErrLine],
-  /// [remainingStdout], and [remainingStderr].
+  /// stream. This stream will then be used for [stdout] and [stderr].
   Stream<String> stdoutStream() {
     var pair = tee(_stdoutLog);
     _stdoutLog = pair.first;
     return pair.last;
   }
 
   /// Returns a stream that will emit anything the process emits via the
   /// process's standard error from now on.
   ///
   /// This stream will be independent from any other methods that deal with
@@ skipping 41 matching lines @@
     schedule(() {
       return _exitCode.then((exitCode) {
         if (expectedExitCode != null) {
           expect(exitCode, equals(expectedExitCode));
         }
       });
     }, "waiting for process '$description' to exit");
     _scheduledExitTask = currentSchedule.tasks.contents.last;
   }
 }