| // 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; |
| |
| /// The queue of tasks to run if an error is caught while running [tasks]. The |
| /// error will be available in [errors]. These tasks won't be run if no error |
| /// occurs. Note that expectation failures count as errors. |
| /// |
| /// This queue runs before [onComplete], and errors in [onComplete] will not |
| /// cause this queue to be run. |
| /// |
| /// If an error occurs in a task in this queue, all further tasks will be |
| /// skipped. |
| TaskQueue get onException => _onException; |
| TaskQueue _onException; |
| |
| /// The queue of tasks to run after [tasks] and possibly [onException] have |
| /// run. This queue will run whether or not an error occurred. If one did, it |
| /// will be available in [errors]. Note that expectation failures count as |
| /// errors. |
| /// |
| /// This queue runs after [onException]. If an error occurs while running |
| /// [onException], that error will be available in [errors] after the original |
| /// error. |
| /// |
| /// If an error occurs in a task in this queue, all further tasks will be |
| /// 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; |
| |
| /// 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. 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. |
| /// |
| /// 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. |
| Duration get timeout => _timeout; |
| Duration _timeout = new Duration(seconds: 30); |
| set timeout(Duration duration) { |
| _timeout = duration; |
| 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(() { |
| 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); |
| 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 (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 (_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) { |
| 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 (_timeout == null) { |
| _timeoutTimer = null; |
| } else { |
| _timeoutTimer = mock_clock.newTimer(_timeout, _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 " |
| "$_timeout 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; |
| |
| _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); |
| }); |
| }).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"); |
| } |
| } |