Add built-in timeouts to 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;
 
+import 'mock_clock.dart' as mock_clock;
 import 'schedule_error.dart';
+import 'substitute_future.dart';
 import 'task.dart';
 
 /// The schedule of tasks to run for a single test. This has three separate task
 /// queues: [tasks], [onComplete], and [onException]. It also provides
 /// visibility into the current state of the schedule.
 class Schedule {
   /// The main task queue for the schedule. These tasks are run before the other
   /// queues and generally constitute the main test body.
   TaskQueue get tasks => _tasks;
   TaskQueue _tasks;
@@ skipping 23 matching lines @@
   /// skipped.
   TaskQueue get onComplete => _onComplete;
   TaskQueue _onComplete;
 
   /// Returns the [Task] that's currently executing, or `null` if there is no
   /// such task. This will be `null` both before the schedule starts running and
   /// after it's finished.
   Task get currentTask => _currentTask;
   Task _currentTask;
 
-  /// Whether the schedule has finished running. This is only set once
-  /// [onComplete] has finished running. It will be set whether or not an
-  /// exception has occurred.
-  bool get done => _done;
-  bool _done = false;
+  /// 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>[];
 
-  /// The task queue that's currently being run, or `null` if there is no such
-  /// queue. One of [tasks], [onException], or [onComplete]. This will be `null`
-  /// before the schedule starts running.
-  TaskQueue get currentQueue => _done ? null : _currentQueue;
+  /// 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, in milliseconds, to wait before terminating a task queue for
+  /// inactivity. Defaults to 30 seconds. This can be set to `null` to disable
+  /// timeouts entirely.
+  ///
+  /// If a task queue times out, an error will be raised that can be handled as
+  /// usual in the [onException] and [onComplete] queues. If [onException] times
+  /// out, that can only be handled in [onComplete]; if [onComplete] times out,
+  /// that cannot be handled.
+  ///
+  /// If a task times out and then later completes with an error, that error
+  /// cannot be handled. The user will still be notified of it.
+  int get timeoutMs => _timeoutMs;
+  int _timeoutMs = 30 * 1000;
+  set timeoutMs(int value) {
+    _timeoutMs = value;
+    heartbeat();
+  }
+
   /// The number of out-of-band callbacks that have been registered with
   /// [wrapAsync] but have yet to be called.
   int _pendingCallbacks = 0;
 
   /// A completer that will be completed once [_pendingCallbacks] reaches zero.
   /// This will only be non-`null` if [_awaitPendingCallbacks] has been called
   /// while [_pendingCallbacks] is non-zero.
   Completer _noPendingCallbacks;
 
+  /// The timer for keeping track of task timeouts. This may be null.
+  Timer _timeoutTimer;
+
   /// Creates a new schedule with empty task queues.
   Schedule() {
     _tasks = new TaskQueue._("tasks", this);
     _onComplete = new TaskQueue._("onComplete", this);
     _onException = new TaskQueue._("onException", this);
+    _currentQueue = _tasks;
+
+    heartbeat();
   }
 
   /// Sets up this schedule by running [setUp], then runs all the task queues in
   /// order. Any errors in [setUp] will cause [onException] to run.
   Future run(void setUp()) {
     return new Future.immediate(null).then((_) {
       try {
         setUp();
       } catch (e, stackTrace) {
         throw new ScheduleError.from(this, e, stackTrace: stackTrace);
       }
 
+      _state = ScheduleState.RUNNING;
       return tasks._run();
     }).catchError((e) {
       errors.add(e);
       return onException._run().catchError((innerError) {
         // If an error occurs in a task in the onException queue, make sure it's
         // registered in the error list and re-throw it. We could also re-throw
         // `e`; ultimately, all the errors will be shown to the user if any
         // ScheduleError is thrown.
         errors.add(innerError);
         throw innerError;
       }).then((_) {
         // If there are no errors in the onException queue, re-throw the
         // original error that caused it to run.
         throw e;
       });
     }).whenComplete(() {
       return onComplete._run().catchError((e) {
         // If an error occurs in a task in the onComplete queue, make sure it's
         // registered in the error list and re-throw it.
         errors.add(e);
         throw e;
       });
     }).whenComplete(() {
-      _done = true;
+      if (_timeoutTimer != null) _timeoutTimer.cancel();
+      _state = ScheduleState.DONE;
     });
   }
 
   /// Signals that an out-of-band error has occurred. Using [wrapAsync] along
   /// with `throw` is usually preferable to calling this directly.
   ///
   /// The metadata in [AsyncError]s and [ScheduleError]s will be preserved.
   void signalError(error, [stackTrace]) {
+    heartbeat();
+
     var scheduleError = new ScheduleError.from(this, error,
-        stackTrace: stackTrace, task: currentTask);
-    if (_done) {
-      errors.add(scheduleError);
+        stackTrace: stackTrace);
+    if (_state == ScheduleState.DONE) {
       throw new StateError(
           "An out-of-band error was signaled outside of wrapAsync after the "
               "schedule finished running.\n"
           "${errorString()}");
-    } else if (currentQueue == null) {
-      // If we're not done but there's no current queue, that means we haven't
-      // started yet and thus we're in setUp or the synchronous body of the
-      // function. Throwing the error will thus pipe it into the main
+    } 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);
     }
   }
 
+  /// 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);
+    errors.add(scheduleError);
+    if (_state == ScheduleState.DONE) {
+      throw new StateError(
+        "An out-of-band error was caught after the test timed out.\n"
+        "${errorString()}");
+    }
+  }
+
   /// Returns a function wrapping [fn] that pipes any errors into the schedule
   /// chain. This will also block the current task queue from completing until
   /// the returned function has been called. It's used to ensure that
   /// out-of-band callbacks are properly handled by the scheduled test.
   ///
   /// The top-level `wrapAsync` function should usually be used in preference to
   /// this.
   Function wrapAsync(fn(arg)) {
-    if (_done) {
+    if (_state == ScheduleState.DONE) {
       throw new StateError("wrapAsync called after the schedule has finished "
           "running.");
     }
+    heartbeat();
+
+    var queue = currentQueue;
+    // It's possible that the queue timed out before this.
+    bool _timedOut() => queue != currentQueue || _pendingCallbacks == 0;
 
     _pendingCallbacks++;
     return (arg) {
       try {
         return fn(arg);
       } catch (e, stackTrace) {
-        signalError(e, stackTrace);
+        if (_timedOut()) {
+          _signalPostTimeoutError(e, stackTrace);
+        } else {
+          signalError(e, stackTrace);
+        }
       } finally {
+        if (_timedOut()) return;
+
         _pendingCallbacks--;
         if (_pendingCallbacks == 0 && _noPendingCallbacks != null) {
           _noPendingCallbacks.complete();
           _noPendingCallbacks = null;
         }
       }
     };
   }
 
   /// 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";
   }
 
+  /// 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 (_timeoutMs == null) {
+      _timeoutTimer = null;
+    } else {
+      _timeoutTimer = mock_clock.getTimer(_timeoutMs, _signalTimeout);
+    }
+  }
+
+  /// The callback to run when the timeout timer fires. Notifies the current
+  /// queue that a timeout has occurred.
+  void _signalTimeout(_) {
+    // Reset the timer so that we can detect timeouts in the onException and
+    // onComplete queues.
+    _timeoutTimer = null;
+
+    var error = new ScheduleError.from(this, "The schedule timed out after "
+        "${_timeoutMs}ms of inactivity.");
+
+    _pendingCallbacks = 0;
+    if (_noPendingCallbacks != null) {
+      var noPendingCallbacks = _noPendingCallbacks;
+      _noPendingCallbacks = null;
+      noPendingCallbacks.completeError(error);
+    } else {
+      currentQueue._signalTimeout(error);
+    }
+  }
+
   /// Returns a [Future] that will complete once there are no pending
   /// out-of-band callbacks.
   Future _awaitNoPendingCallbacks() {
     if (_pendingCallbacks == 0) return new Future.immediate(null);
     if (_noPendingCallbacks == null) _noPendingCallbacks = new Completer();
     return _noPendingCallbacks.future;
   }
 }
 
+/// An enum of states for a [Schedule].
+class ScheduleState {
+  /// The schedule can have tasks added to its queue, but is not yet running
+  /// them.
+  static const SET_UP = const ScheduleState._("SET_UP");
+
+  /// The schedule is actively running tasks. This includes running tasks in
+  /// [Schedule.onException] and [Schedule.onComplete].
+  static const RUNNING = const ScheduleState._("RUNNING");
+
+  /// The schedule has finished running all its tasks, either successfully or
+  /// with an error.
+  static const DONE = const ScheduleState._("DONE");
+
+  /// The name of the state.
+  final String name;
+
+  const ScheduleState._(this.name);
+
+  String toString() => name;
+}
+
 /// A queue of asynchronous tasks to execute in order.
 class TaskQueue {
   // TODO(nweiz): make this a read-only view when issue 8321 is fixed.
   /// The tasks in the queue.
   Collection<Task> get contents => _contents;
   final _contents = new Queue<Task>();
 
   /// The name of the queue, for debugging purposes.
   final String name;
 
   /// The [Schedule] that created this queue.
   final Schedule _schedule;
 
   /// 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);
 
   /// 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
   /// running. Its return value is the same as the return value of [fn], or the
   /// value it completes to if it's a [Future].
   ///
   /// If [description] is passed, it's used to describe the task for debugging
   /// purposes when an error occurs.
   Future schedule(fn(), [String description]) {
     var task = new Task(fn, this, description);
     _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;
-      return task.fn().catchError((e) {
+
+      _taskFuture = new SubstituteFuture(task.fn());
+      return _taskFuture.whenComplete(() {
+        _taskFuture = null;
+        _schedule.heartbeat();
+      }).catchError((e) {
         if (_error != null) _schedule.errors.add(_error);
-        throw new ScheduleError.from(_schedule, e, task: task);
+        throw new ScheduleError.from(_schedule, e);
       });
     }).whenComplete(() {
       _schedule._currentTask = null;
       return _schedule._awaitNoPendingCallbacks();
     }).then((_) {
+      _schedule.heartbeat();
       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.errors.add(_error);
     _error = error;
   }
 
+  /// Notifies the queue that it has timed out and it needs to terminate
+  /// immediately with a timeout error.
+  void _signalTimeout(ScheduleError error) {
+    if (_taskFuture != null) {
+      // Catch errors coming off the old task future, in case it completes after
+      // timing out.
+      _taskFuture.substitute(new Future.immediateError(error)).catchError((e) {
+        _schedule._signalPostTimeoutError(e);
+      });
+    } else {
+      // This branch probably won't be reached, but it's conceivable that the
+      // event loop might get pumped when _taskFuture is null but we haven't yet
+      // called _awaitNoPendingCallbacks.
+      _signalError(error);
+    }
+  }
+
   String toString() => name;
 
   /// Returns a detailed representation of the queue as a tree of tasks. If
   /// [highlight] is passed, that task is specially highlighted.
   ///
   /// [highlight] must be a task in this queue.
   String generateTree([Task highlight]) {
     assert(highlight == null || highlight.queue == this);
     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");
   }
 }