Add a ScheduledProcess class to pkg/scheduled_test.

pkg/scheduled_test/lib/src/task.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 task;
 
 import 'dart:async';
 import 'dart:collection';
 
 import 'future_group.dart';
@@ skipping 26 matching lines @@
   /// A description of this task. Used for debugging. May be `null`.
   final String description;
 
   /// The parent task, if this is a nested task that was started while another
   /// task was running. This will be `null` for top-level tasks.
   final Task parent;
 
   /// The body of the task.
   TaskBody fn;
 
+  /// The current state of [this].
+  TaskState get state => _state;
+  var _state = TaskState.WAITING;
+
   /// The identifier of the task. For top-level tasks, this is the index of the
   /// task within [queue]; for nested tasks, this is the index within
   /// [parent.children]. It's used for debugging when [description] isn't
   /// provided.
   int _id;
 
   /// A Future that will complete to the return value of [fn] once this task
   /// finishes running.
   Future get result => _resultCompleter.future;
   final _resultCompleter = new Completer();
 
   Task(fn(), String description, TaskQueue queue)
     : this._(fn, description, queue, null, queue.contents.length);
 
   Task._child(fn(), String description, Task parent)
     : this._(fn, description, parent.queue, parent, parent.children.length);
 
   Task._(fn(), this.description, this.queue, this.parent, this._id) {
     this.fn = () {
+      if (state != TaskState.WAITING) {
+        throw new StateError("Can't run $state task '$this'.");
+      }
+
+      _state = TaskState.RUNNING;
       var future = new Future.immediate(null).then((_) => fn())
           .whenComplete(() {
         if (_childGroup == null || _childGroup.completed) return;
         return _childGroup.future;
       });
       chainToCompleter(future, _resultCompleter);
       return future;
     };
 
-    // Make sure any error thrown by fn isn't top-leveled by virtue of being
-    // passed to the result future.
-    result.catchError((_) {});
+    // If the parent queue experiences an error before this task has started
+    // running, pipe that error out through [result]. This ensures that we don't
+    // get deadlocked by something like `expect(schedule(...), completes)`.
+    queue.onTasksComplete.catchError((e) {
+      if (state == TaskState.WAITING) _resultCompleter.completeError(e);
+    });
+
+    // catchError makes sure any error thrown by fn isn't top-leveled by virtue
+    // of being passed to the result future.
+    result.whenComplete(() {
+      _state = TaskState.DONE;
+    }).catchError((_) {});
   }
 
   /// Run [fn] as a child of this task. Returns a Future that will complete with
   /// the result of the child task. This task will not complete until [fn] has
   /// finished.
   Future runChild(fn(), String description) {
     var task = new Task._child(fn, description, this);
     children.add(task);
     if (_childGroup == null || _childGroup.completed) {
       _childGroup = new FutureGroup();
     }
     // Ignore errors in the FutureGroup; they'll get picked up via wrapFuture,
     // and we don't want them to short-circuit the other Futures.
     _childGroup.add(task.result.catchError((_) {}));
     task.fn();
     return task.result;
   }
 
   String toString() => description == null ? "#$_id" : description;
 
   /// Returns a detailed representation of [queue] with this task highlighted.
   String generateTree() => queue.generateTree(this);
 }
+
+/// An enum of states for a [Task].
+class TaskState {
+  /// The task is waiting to be run.
+  static const WAITING = const TaskState._("WAITING");
+
+  /// The task is currently running.
+  static const RUNNING = const TaskState._("RUNNING");
+
+  /// The task has finished running, either successfully or with an error.
+  static const DONE = const TaskState._("DONE");
+
+  /// The name of the state.
+  final String name;
+
+  const TaskState._(this.name);
+
+  String toString() => name;
+}