Accommodate a race condition in pkg/scheduled_test/scheduled_process_test.

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_process;
 
 import 'dart:async';
 import 'dart:io';
 
 import 'scheduled_test.dart';
@@ skipping 20 matching lines @@
   /// 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;
 
-  /// A subscription that controls both [_stdout] and [_stdoutLog].
-  StreamSubscription<String> _stdoutSubscription;
+  /// 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;
 
-  /// A subscription that controls both [_stderr] and [_stderrLog].
-  StreamSubscription<String> _stderrSubscription;
+  /// 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].
   var _endScheduled = false;
 
   /// The task that runs immediately before this process is scheduled to end. If
@@ skipping 17 matching lines @@
       {options, String description, Encoding encoding: Encoding.UTF_8})
       : _encoding = encoding {
     assert(currentSchedule.state == ScheduleState.SET_UP);
 
     _updateDescription(executable, arguments);
 
     _scheduleStartProcess(executable, arguments, options);
 
     _scheduleExceptionCleanup();
 
-    var stdoutWithSubscription = _lineStreamWithSubscription(
+    var stdoutWithCanceller = _lineStreamWithCanceller(
         _process.then((p) => p.stdout));
-    _stdoutSubscription = stdoutWithSubscription.last;
-    var stdoutTee = tee(stdoutWithSubscription.first);
+    _stdoutCanceller = stdoutWithCanceller.last;
+    var stdoutTee = tee(stdoutWithCanceller.first);
     _stdout = stdoutTee.first;
     _stdoutLog = stdoutTee.last;
 
-    var stderrWithSubscription = _lineStreamWithSubscription(
+    var stderrWithCanceller = _lineStreamWithCanceller(
         _process.then((p) => p.stderr));
-    _stderrSubscription = stderrWithSubscription.last;
-    var stderrTee = tee(stderrWithSubscription.first);
+    _stderrCanceller = stderrWithCanceller.last;
+    var stderrTee = tee(stderrWithCanceller.first);
     _stderr = stderrTee.first;
     _stderrLog = stderrTee.last;
   }
 
   /// Updates [_description] to reflect [executable] and [arguments], which are
   /// the same values as in [start].
   void _updateDescription(executable, arguments) {
     if (executable is Future) {
       _description = "future process";
     } else if (arguments is Future || arguments.any((e) => e is Future)) {
@@ skipping 59 matching lines @@
 
         if (!_endExpected) {
           throw "Process '${this.description}' ended earlier than scheduled "
             "with exit code $exitCode.";
         }
       }));
     });
   }
 
   /// Converts a stream of bytes to a stream of lines and returns that along
-  /// with a [StreamSubscription] controlling it.
-  Pair<Stream<String>, StreamSubscription<String>> _lineStreamWithSubscription(
+  /// with a [StreamCanceller] controlling it.
+  Pair<Stream<String>, StreamCanceller> _lineStreamWithCanceller(
       Future<Stream<int>> streamFuture) {
-    return streamWithSubscription(futureStream(streamFuture)
+    return streamWithCanceller(futureStream(streamFuture)
         .handleError((e) => currentSchedule.signalError(e))
         .transform(new StringDecoder(_encoding))
         .transform(new LineTransformer()));
   }
 
   /// Schedule an exception handler that will clean up the process and provide
   /// debug information if an error occurs.
   void _scheduleExceptionCleanup() {
     currentSchedule.onException.schedule(() {
-      _stdoutSubscription.cancel();
-      _stderrSubscription.cancel();
+      _stdoutCanceller();
+      _stderrCanceller();
 
       if (!_process.hasValue) return;
 
       var killedPrematurely = false;
       if (!_exitCode.hasValue) {
         killedPrematurely = true;
         _endExpected = true;
         _process.value.kill();
         // Ensure that the onException queue waits for the process to actually
         // exit after being killed.
@@ skipping 92 matching lines @@
     schedule(() {
       _endExpected = true;
       return _exitCode.then((exitCode) {
         if (expectedExitCode != null) {
           expect(exitCode, equals(expectedExitCode));
         }
       });
     }, "waiting for process '$description' to exit");
   }
 }