Add a ScheduledProcess class to pkg/scheduled_test.

pkg/scheduled_test/lib/src/schedule.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 schedule;
 
 import 'dart:async';
 import 'dart:collection';
 
 import 'package:unittest/unittest.dart' as unittest;
@@ skipping 42 matching lines @@
   /// such task. This will be `null` both before the schedule starts running and
   /// after it's finished.
   Task get currentTask => _currentTask;
   Task _currentTask;
 
   /// The current state of the schedule.
   ScheduleState get state => _state;
   ScheduleState _state = ScheduleState.SET_UP;
 
   // TODO(nweiz): make this a read-only view once issue 8321 is fixed.
-
   /// Errors thrown by the task queues.
   ///
   /// When running tasks in [tasks], this will always be empty. If an error
   /// occurs in [tasks], it will be added to this list and then [onException]
   /// will be run. If an error occurs there as well, it will be added to this
   /// list and [onComplete] will be run. Errors thrown during [onComplete] will
   /// also be added to this list, although no scheduled tasks will be run
   /// afterwards.
   ///
   /// Any out-of-band callbacks that throw errors will also have those errors
   /// added to this list.
   final errors = <ScheduleError>[];
 
+  // TODO(nweiz): make this a read-only view once issue 8321 is fixed.
+  /// Additional debugging info registered via [addDebugInfo].
+  final debugInfo = <String>[];
+
   /// The task queue that's currently being run. One of [tasks], [onException],
   /// or [onComplete]. This starts as [tasks], and can only be `null` after the
   /// schedule is done.
   TaskQueue get currentQueue =>
     _state == ScheduleState.DONE ? null : _currentQueue;
   TaskQueue _currentQueue;
 
   /// The time to wait before terminating a task queue for inactivity. Defaults
   /// to 30 seconds. This can be set to `null` to disable timeouts entirely.
   ///
@@ skipping 91 matching lines @@
           "${errorString()}");
     } else if (state == ScheduleState.SET_UP) {
       // If we're setting up, throwing the error will pipe it into the main
       // error-handling code.
       throw scheduleError;
     } else {
       _currentQueue._signalError(scheduleError);
     }
   }
 
+  /// Adds [info] to the debugging output that will be printed if the test
+  /// fails. Unlike [signalError], this won't cause the test to fail, nor will
+  /// it short-circuit the current [TaskQueue]; it's just useful for providing
+  /// additional information that may not fit cleanly into an existing error.
+  void addDebugInfo(String info) => debugInfo.add(info);
+
   /// Notifies the schedule of an error that occurred in a task or out-of-band
   /// callback after the appropriate queue has timed out. If this schedule is
   /// still running, the error will be added to the errors list to be shown
   /// along with the timeout error; otherwise, a top-level error will be thrown.
   void _signalPostTimeoutError(error, [stackTrace]) {
     var scheduleError = new ScheduleError.from(this, error,
         stackTrace: stackTrace);
     _addError(scheduleError);
     if (_state == ScheduleState.DONE) {
       throw new StateError(
@@ skipping 67 matching lines @@
     // Don't top-level the error, since it's already been signaled to the
     // schedule.
     future.catchError((_) => null);
 
     return future;
   }
 
   /// Returns a string representation of all errors registered on this schedule.
   String errorString() {
     if (errors.isEmpty) return "The schedule had no errors.";
-    if (errors.length == 1) return errors.first.toString();
-    var errorStrings = errors.map((e) => e.toString()).join("\n================"
-        "================================================================\n");
-    return "The schedule had ${errors.length} errors:\n$errorStrings";
+    if (errors.length == 1 && debugInfo.isEmpty) return errors.first.toString();
+
+    var border = "\n==========================================================="
+      "=====================\n";
+    var errorStrings = errors.map((e) => e.toString()).join(border);
+    var message = "The schedule had ${errors.length} errors:\n$errorStrings";
+
+    if (!debugInfo.isEmpty) {
+      message = "$message$border\nDebug info:\n${debugInfo.join(border)}";
+    }
+
+    return message;
   }
 
   /// Notifies the schedule that progress is being made on an asynchronous task.
   /// This resets the timeout timer, and can be used in long-running tasks to
   /// keep them from timing out.
   void heartbeat() {
     if (_timeoutTimer != null) _timeoutTimer.cancel();
     if (_timeout == null) {
       _timeoutTimer = null;
     } else {
@@ skipping 75 matching lines @@
 
   /// An out-of-band error signaled by [_schedule]. If this is non-null, it
   /// indicates that the queue should stop as soon as possible and re-throw this
   /// error.
   ScheduleError _error;
 
   /// The [SubstituteFuture] for the currently-running task in the queue, or
   /// null if no task is currently running.
   SubstituteFuture _taskFuture;
 
-  TaskQueue._(this.name, this._schedule);
+  /// A [Future] that completes when the tasks in [this] are all complete. If an
+  /// error occurs while running this queue, the returned [Future] will complete
+  /// with that error.
+  ///
+  /// The returned [Future] can complete before outstanding out-of-band
+  /// callbacks have finished running.
+  Future get onTasksComplete => _onTasksCompleteCompleter.future;
+  final _onTasksCompleteCompleter = new Completer();
+
+  TaskQueue._(this.name, this._schedule) {
+    // Avoid top-leveling errors that are passed to onTasksComplete if there are no

[Bob Nystrom, 2013/03/04 23:52:00]

Long line.

[nweiz, 2013/03/05 02:16:09]

Done.

+    // listeners.
+    onTasksComplete.catchError((_) {});
+  }
 
   /// Whether this queue is currently running.
   bool get isRunning => _schedule.state == ScheduleState.RUNNING &&
       _schedule.currentQueue == this;
 
   /// Schedules a task, [fn], to run asynchronously as part of this queue. Tasks
   /// will be run in the order they're scheduled. In [fn] returns a [Future],
   /// tasks after it won't be run until that [Future] completes.
   ///
   /// The return value will be completed once the scheduled task has finished
@@ skipping 11 matching lines @@
   /// top-level tasks which run in sequence.
   Future schedule(fn(), [String description]) {
     if (isRunning) {
       var task = _schedule.currentTask;
       var wrappedFn = () => _schedule.wrapFuture(
           new Future.immediate(null).then((_) => fn()));
       if (task == null) return wrappedFn();
       return task.runChild(wrappedFn, description);
     }
 
-    var task = new Task(fn, description, this);
+    var task = new Task(() {
+      return new Future.of(fn).catchError((e) {
+        throw new ScheduleError.from(_schedule, e);
+      });
+    }, description, this);
     _contents.add(task);
     return task.result;
   }
 
   /// Runs all the tasks in this queue in order.
   Future _run() {
     _schedule._currentQueue = this;
     _schedule.heartbeat();
     return Future.forEach(_contents, (task) {
       _schedule._currentTask = task;
       if (_error != null) throw _error;
 
       _taskFuture = new SubstituteFuture(task.fn());
       return _taskFuture.whenComplete(() {
         _taskFuture = null;
         _schedule.heartbeat();
       }).catchError((e) {
-        if (_error != null) _schedule._addError(_error);
-        throw new ScheduleError.from(_schedule, e);
+        var error = new ScheduleError.from(_schedule, e);
+        _signalError(error);
+        throw _error;
+      });
+    }).then((_) {
+      _onTasksCompleteCompleter.complete();
+    }).catchError((e) {
+      _onTasksCompleteCompleter.completeError(e);
+      throw e;
+    }).whenComplete(() {
+      _schedule._currentTask = null;
+      return _schedule._awaitNoPendingCallbacks().catchError((e) {
+        // Signal the error rather than passing it through directly so that if a
+        // timeout happens after an in-task error, both are reported.
+        _signalError(new ScheduleError.from(_schedule, e));
       });
     }).whenComplete(() {
-      _schedule._currentTask = null;
-      return _schedule._awaitNoPendingCallbacks();
-    }).then((_) {
       _schedule.heartbeat();
+      // If the tasks were otherwise successful, make sure we throw any
+      // out-of-band errors. If a task failed, make sure we throw the most
+      // recent error.
       if (_error != null) throw _error;
     });
   }
 
   /// Signals that an out-of-band error has been detected and the queue should
   /// stop running as soon as possible.
   void _signalError(ScheduleError error) {
     // If multiple errors are detected while a task is running, make sure the
     // earlier ones are recorded in the schedule.
     if (_error != null) _schedule._addError(_error);
@@ skipping 28 matching lines @@
     return _contents.map((task) {
       var lines = task.toString().split("\n");
       var firstLine = task == highlight ?
           "> ${lines.first}" : "* ${lines.first}";
       lines = new List.from(lines.skip(1).map((line) => "| $line"));
       lines.insertRange(0, 1, firstLine);
       return lines.join("\n");
     }).join("\n");
   }
 }