| OLD | NEW |
| 1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file | 1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file |
| 2 // for details. All rights reserved. Use of this source code is governed by a | 2 // for details. All rights reserved. Use of this source code is governed by a |
| 3 // BSD-style license that can be found in the LICENSE file. | 3 // BSD-style license that can be found in the LICENSE file. |
| 4 | 4 |
| 5 part of dart.async; | 5 part of dart.async; |
| 6 | 6 |
| 7 /** The onValue and onError handlers return either a value or a future */ | 7 abstract class _Completer<T> implements Completer<T> { |
| 8 typedef dynamic _FutureOnValue<T>(T value); | 8 final Future<T> future; |
| 9 typedef dynamic _FutureOnError(error); | 9 bool _isComplete = false; |
| 10 /** Test used by [Future.catchError] to handle skip some errors. */ | |
| 11 typedef bool _FutureErrorTest(var error); | |
| 12 /** Used by [WhenFuture]. */ | |
| 13 typedef _FutureAction(); | |
| 14 | 10 |
| 15 abstract class _Completer<T> implements Completer<T> { | 11 _Completer() : future = new _FutureImpl<T>() { |
| 16 final _Future<T> future = new _Future<T>(); | 12 _FutureImpl futureImpl = future; |
| 13 futureImpl._zone.expectCallback(); |
| 14 } |
| 17 | 15 |
| 18 void complete([T value]); | 16 void _setFutureValue(T value); |
| 17 void _setFutureError(error); |
| 19 | 18 |
| 20 void completeError(Object error, [Object stackTrace = null]); | 19 void complete([T value]) { |
| 20 if (_isComplete) throw new StateError("Future already completed"); |
| 21 _isComplete = true; |
| 22 _FutureImpl futureImpl = future; |
| 23 _setFutureValue(value); |
| 24 } |
| 21 | 25 |
| 22 // The future's _isComplete doesn't take into account pending completions. | 26 void completeError(Object error, [Object stackTrace = null]) { |
| 23 // We therefore use _mayComplete. | 27 if (_isComplete) throw new StateError("Future already completed"); |
| 24 bool get isCompleted => !future._mayComplete; | 28 _isComplete = true; |
| 29 if (stackTrace != null) { |
| 30 // Force the stack trace onto the error, even if it already had one. |
| 31 _attachStackTrace(error, stackTrace); |
| 32 } |
| 33 _FutureImpl futureImpl = future; |
| 34 _setFutureError(error); |
| 35 } |
| 36 |
| 37 bool get isCompleted => _isComplete; |
| 25 } | 38 } |
| 26 | 39 |
| 27 class _AsyncCompleter<T> extends _Completer<T> { | 40 class _AsyncCompleter<T> extends _Completer<T> { |
| 28 | 41 void _setFutureValue(T value) { |
| 29 void complete([T value]) { | 42 _FutureImpl future = this.future; |
| 30 future._asyncComplete(value); | 43 future._asyncSetValue(value); |
| 44 // The async-error will schedule another callback, so we can cancel |
| 45 // the expectation without shutting down the zone. |
| 46 future._zone.cancelCallbackExpectation(); |
| 31 } | 47 } |
| 32 | 48 |
| 33 void completeError(Object error, [Object stackTrace = null]) { | 49 void _setFutureError(error) { |
| 34 future._asyncCompleteError(error, stackTrace); | 50 _FutureImpl future = this.future; |
| 51 future._asyncSetError(error); |
| 52 // The async-error will schedule another callback, so we can cancel |
| 53 // the expectation without shutting down the zone. |
| 54 future._zone.cancelCallbackExpectation(); |
| 35 } | 55 } |
| 36 } | 56 } |
| 37 | 57 |
| 38 class _SyncCompleter<T> extends _Completer<T> { | 58 class _SyncCompleter<T> extends _Completer<T> { |
| 39 | 59 void _setFutureValue(T value) { |
| 40 void complete([T value]) { | 60 _FutureImpl future = this.future; |
| 41 future._complete(value); | 61 future._setValue(value); |
| 62 future._zone.cancelCallbackExpectation(); |
| 42 } | 63 } |
| 43 | 64 |
| 44 void completeError(Object error, [Object stackTrace = null]) { | 65 void _setFutureError(error) { |
| 45 future._completeError(error, stackTrace); | 66 _FutureImpl future = this.future; |
| 67 future._setError(error); |
| 68 future._zone.cancelCallbackExpectation(); |
| 46 } | 69 } |
| 47 } | 70 } |
| 48 | 71 |
| 49 class _Future<T> implements Future<T> { | 72 /** |
| 73 * A listener on a future. |
| 74 * |
| 75 * When the future completes, the [_sendValue] or [_sendError] method |
| 76 * is invoked with the result. |
| 77 * |
| 78 * Listeners are kept in a linked list. |
| 79 */ |
| 80 abstract class _FutureListener<T> { |
| 81 _FutureListener _nextListener; |
| 82 factory _FutureListener.wrap(_FutureImpl future) { |
| 83 return new _FutureListenerWrapper(future); |
| 84 } |
| 85 void _sendValue(T value); |
| 86 void _sendError(error); |
| 87 |
| 88 bool _inSameErrorZone(_Zone otherZone); |
| 89 } |
| 90 |
| 91 /** Adapter for a [_FutureImpl] to be a future result listener. */ |
| 92 class _FutureListenerWrapper<T> implements _FutureListener<T> { |
| 93 _FutureImpl future; |
| 94 _FutureListener _nextListener; |
| 95 _FutureListenerWrapper(this.future); |
| 96 _sendValue(T value) { future._setValueUnchecked(value); } |
| 97 _sendError(error) { future._setErrorUnchecked(error); } |
| 98 bool _inSameErrorZone(_Zone otherZone) => future._inSameErrorZone(otherZone); |
| 99 } |
| 100 |
| 101 /** |
| 102 * This listener is installed at error-zone boundaries. It signals an |
| 103 * uncaught error in the zone of origin when an error is sent from one error |
| 104 * zone to another. |
| 105 * |
| 106 * When a Future is listening to another Future and they have not been |
| 107 * instantiated in the same error-zone then Futures put an instance of this |
| 108 * class between them (see [_FutureImpl._addListener]). |
| 109 * |
| 110 * For example: |
| 111 * |
| 112 * var completer = new Completer(); |
| 113 * var future = completer.future.then((x) => x); |
| 114 * catchErrors(() { |
| 115 * var future2 = future.catchError(print); |
| 116 * }); |
| 117 * completer.completeError(499); |
| 118 * |
| 119 * In this example `future` and `future2` are in different error-zones. The |
| 120 * error (499) that originates outside `catchErrors` must not reach the |
| 121 * `catchError` future (`future2`) inside `catchErrors`. |
| 122 * |
| 123 * When invoking `catchError` on `future` the Future installs an |
| 124 * [_ErrorZoneBoundaryListener] between itself and the result, `future2`. |
| 125 * |
| 126 * Conceptually _ErrorZoneBoundaryListeners could be implemented as |
| 127 * `catchError`s on the origin future as well. |
| 128 */ |
| 129 class _ErrorZoneBoundaryListener implements _FutureListener { |
| 130 _FutureListener _nextListener; |
| 131 final _FutureListener _listener; |
| 132 |
| 133 _ErrorZoneBoundaryListener(this._listener); |
| 134 |
| 135 bool _inSameErrorZone(_Zone otherZone) { |
| 136 // Should never be called. We use [_inSameErrorZone] to know if we have |
| 137 // to insert an instance of [_ErrorZoneBoundaryListener] (and in the |
| 138 // controller). Once we have inserted one we should never need to use it |
| 139 // anymore. |
| 140 throw new UnsupportedError( |
| 141 "A Zone boundary doesn't support the inSameErrorZone test."); |
| 142 } |
| 143 |
| 144 void _sendValue(value) { |
| 145 _listener._sendValue(value); |
| 146 } |
| 147 |
| 148 void _sendError(error) { |
| 149 // We are not allowed to send an error from one error-zone to another. |
| 150 // This is the whole purpose of this class. |
| 151 _Zone.current.handleUncaughtError(error); |
| 152 } |
| 153 } |
| 154 |
| 155 class _FutureImpl<T> implements Future<T> { |
| 50 // State of the future. The state determines the interpretation of the | 156 // State of the future. The state determines the interpretation of the |
| 51 // [resultOrListeners] field. | 157 // [resultOrListeners] field. |
| 52 // TODO(lrn): rename field since it can also contain a chained future. | 158 // TODO(lrn): rename field since it can also contain a chained future. |
| 53 | 159 |
| 54 /// Initial state, waiting for a result. In this state, the | 160 /// Initial state, waiting for a result. In this state, the |
| 55 /// [resultOrListeners] field holds a single-linked list of | 161 /// [resultOrListeners] field holds a single-linked list of |
| 56 /// [FutureListener] listeners. | 162 /// [FutureListener] listeners. |
| 57 static const int _INCOMPLETE = 0; | 163 static const int _INCOMPLETE = 0; |
| 58 /// Pending completion. Set when completed using [_asyncComplete] or | 164 /// Pending completion. Set when completed using [_asyncSetValue] or |
| 59 /// [_asyncCompleteError]. It is an error to try to complete it again. | 165 /// [_asyncSetError]. It is an error to try to complete it again. |
| 60 static const int _PENDING_COMPLETE = 1; | 166 static const int _PENDING_COMPLETE = 1; |
| 61 /// The future has been chained to another future. The result of that | 167 /// The future has been chained to another future. The result of that |
| 62 /// other future becomes the result of this future as well. | 168 /// other future becomes the result of this future as well. |
| 63 /// In this state, no callback should be executed anymore. | 169 /// In this state, the [resultOrListeners] field holds the future that |
| 64 // TODO(floitsch): we don't really need a special "_CHAINED" state. We could | 170 /// will give the result to this future. Both existing and new listeners are |
| 65 // just use the PENDING_COMPLETE state instead. | 171 /// forwarded directly to the other future. |
| 66 static const int _CHAINED = 2; | 172 static const int _CHAINED = 2; |
| 173 /// The future has been chained to another future, but there hasn't been |
| 174 /// any listeners added to this future yet. If it is completed with an |
| 175 /// error, the error will be considered unhandled. |
| 176 static const int _CHAINED_UNLISTENED = 6; |
| 67 /// The future has been completed with a value result. | 177 /// The future has been completed with a value result. |
| 68 static const int _VALUE = 4; | 178 static const int _VALUE = 8; |
| 69 /// The future has been completed with an error result. | 179 /// The future has been completed with an error result. |
| 70 static const int _ERROR = 8; | 180 static const int _ERROR = 12; |
| 71 | 181 |
| 72 /** Whether the future is complete, and as what. */ | 182 /** Whether the future is complete, and as what. */ |
| 73 int _state = _INCOMPLETE; | 183 int _state = _INCOMPLETE; |
| 74 | 184 |
| 75 final _Zone _zone = _Zone.current.fork(); | 185 final _Zone _zone = _Zone.current.fork(); |
| 76 | 186 |
| 187 bool get _isChained => (_state & _CHAINED) != 0; |
| 188 bool get _hasChainedListener => _state == _CHAINED; |
| 189 bool get _isComplete => _state >= _VALUE; |
| 77 bool get _mayComplete => _state == _INCOMPLETE; | 190 bool get _mayComplete => _state == _INCOMPLETE; |
| 78 bool get _isChained => _state == _CHAINED; | |
| 79 bool get _isComplete => _state >= _VALUE; | |
| 80 bool get _hasValue => _state == _VALUE; | 191 bool get _hasValue => _state == _VALUE; |
| 81 bool get _hasError => _state == _ERROR; | 192 bool get _hasError => _state >= _ERROR; |
| 82 | |
| 83 set _isChained(bool value) { | |
| 84 if (value) { | |
| 85 assert(_mayComplete); | |
| 86 _state = _CHAINED; | |
| 87 } else { | |
| 88 assert(_isChained); | |
| 89 _state = _INCOMPLETE; | |
| 90 } | |
| 91 } | |
| 92 | 193 |
| 93 /** | 194 /** |
| 94 * Either the result, a list of listeners or another future. | 195 * Either the result, a list of listeners or another future. |
| 95 * | 196 * |
| 96 * The result of the future is either a value or an error. | 197 * The result of the future is either a value or an error. |
| 97 * A result is only stored when the future has completed. | 198 * A result is only stored when the future has completed. |
| 98 * | 199 * |
| 99 * The listeners is an internally linked list of [_FutureListener]s. | 200 * The listeners is an internally linked list of [_FutureListener]s. |
| 100 * Listeners are only remembered while the future is not yet complete, | 201 * Listeners are only remembered while the future is not yet complete, |
| 101 * and it is not chained to another future. | 202 * and it is not chained to another future. |
| 102 * | 203 * |
| 103 * The future is another future that his future is chained to. This future | 204 * The future is another future that his future is chained to. This future |
| 104 * is waiting for the other future to complete, and when it does, this future | 205 * is waiting for the other future to complete, and when it does, this future |
| 105 * will complete with the same result. | 206 * will complete with the same result. |
| 106 * All listeners are forwarded to the other future. | 207 * All listeners are forwarded to the other future. |
| 107 * | 208 * |
| 108 * The cases are disjoint (incomplete and unchained, incomplete and | 209 * The cases are disjoint (incomplete and unchained, incomplete and |
| 109 * chained, or completed with value or error), so the field only needs to hold | 210 * chained, or completed with value or error), so the field only needs to hold |
| 110 * one value at a time. | 211 * one value at a time. |
| 111 */ | 212 */ |
| 112 var _resultOrListeners; | 213 var _resultOrListeners; |
| 113 | 214 |
| 114 /** | 215 _FutureImpl(); |
| 115 * A [_Future] implements a linked list. If a future has more than one | |
| 116 * listener the [_nextListener] field of the first listener points to the | |
| 117 * remaining listeners. | |
| 118 */ | |
| 119 // TODO(floitsch): since single listeners are the common case we should | |
| 120 // use a bit to indicate that the _resultOrListeners contains a container. | |
| 121 _Future _nextListener; | |
| 122 | 216 |
| 123 // TODO(floitsch): we only need two closure fields to store the callbacks. | 217 _FutureImpl.immediate(T value) { |
| 124 // If we store the type of a closure in the state field (where there are | 218 _state = _VALUE; |
| 125 // still bits left), we can just store two closures instead of using 4 | 219 _resultOrListeners = value; |
| 126 // fields of which 2 are always null. | |
| 127 final _FutureOnValue _onValueCallback; | |
| 128 final _FutureErrorTest _errorTestCallback; | |
| 129 final _FutureOnError _onErrorCallback; | |
| 130 final _FutureAction _whenCompleteActionCallback; | |
| 131 | |
| 132 _FutureOnValue get _onValue => _isChained ? null : _onValueCallback; | |
| 133 _FutureErrorTest get _errorTest => _isChained ? null : _errorTestCallback; | |
| 134 _FutureOnError get _onError => _isChained ? null : _onErrorCallback; | |
| 135 _FutureAction get _whenCompleteAction | |
| 136 => _isChained ? null : _whenCompleteActionCallback; | |
| 137 | |
| 138 _Future() | |
| 139 : _onValueCallback = null, _errorTestCallback = null, | |
| 140 _onErrorCallback = null, _whenCompleteActionCallback = null; | |
| 141 | |
| 142 _Future.immediate(T value) | |
| 143 : _onValueCallback = null, _errorTestCallback = null, | |
| 144 _onErrorCallback = null, _whenCompleteActionCallback = null { | |
| 145 _asyncComplete(value); | |
| 146 } | 220 } |
| 147 | 221 |
| 148 _Future.immediateError(var error, [Object stackTrace]) | 222 _FutureImpl.immediateError(var error, [Object stackTrace]) { |
| 149 : _onValueCallback = null, _errorTestCallback = null, | 223 if (stackTrace != null) { |
| 150 _onErrorCallback = null, _whenCompleteActionCallback = null { | 224 // Force stack trace onto error, even if it had already one. |
| 151 _asyncCompleteError(error, stackTrace); | 225 _attachStackTrace(error, stackTrace); |
| 226 } |
| 227 _asyncSetError(error); |
| 152 } | 228 } |
| 153 | 229 |
| 154 _Future._then(this._onValueCallback, this._onErrorCallback) | 230 factory _FutureImpl.wait(Iterable<Future> futures) { |
| 155 : _errorTestCallback = null, _whenCompleteActionCallback = null { | 231 Completer completer; |
| 156 _zone.expectCallback(); | 232 // List collecting values from the futures. |
| 157 } | 233 // Set to null if an error occurs. |
| 158 | 234 List values; |
| 159 _Future._catchError(this._onErrorCallback, this._errorTestCallback) | 235 void handleError(error) { |
| 160 : _onValueCallback = null, _whenCompleteActionCallback = null { | 236 if (values != null) { |
| 161 _zone.expectCallback(); | 237 values = null; |
| 162 } | 238 completer.completeError(error); |
| 163 | 239 } |
| 164 _Future._whenComplete(this._whenCompleteActionCallback) | 240 } |
| 165 : _onValueCallback = null, _errorTestCallback = null, | 241 // As each future completes, put its value into the corresponding |
| 166 _onErrorCallback = null { | 242 // position in the list of values. |
| 167 _zone.expectCallback(); | 243 int remaining = 0; |
| 244 for (Future future in futures) { |
| 245 int pos = remaining++; |
| 246 future.catchError(handleError).then((Object value) { |
| 247 if (values == null) return null; |
| 248 values[pos] = value; |
| 249 remaining--; |
| 250 if (remaining == 0) { |
| 251 completer.complete(values); |
| 252 } |
| 253 }); |
| 254 } |
| 255 if (remaining == 0) { |
| 256 return new Future.value(const []); |
| 257 } |
| 258 values = new List(remaining); |
| 259 completer = new Completer<List>(); |
| 260 return completer.future; |
| 168 } | 261 } |
| 169 | 262 |
| 170 Future then(f(T value), { onError(error) }) { | 263 Future then(f(T value), { onError(error) }) { |
| 171 _Future result; | 264 if (onError == null) { |
| 172 result = new _Future._then(f, onError); | 265 return new _ThenFuture(f).._subscribeTo(this); |
| 173 _addListener(result); | 266 } |
| 174 return result; | 267 return new _SubscribeFuture(f, onError).._subscribeTo(this); |
| 175 } | 268 } |
| 176 | 269 |
| 177 Future catchError(f(error), { bool test(error) }) { | 270 Future catchError(f(error), { bool test(error) }) { |
| 178 _Future result = new _Future._catchError(f, test); | 271 return new _CatchErrorFuture(f, test).._subscribeTo(this); |
| 179 _addListener(result); | |
| 180 return result; | |
| 181 } | 272 } |
| 182 | 273 |
| 183 Future<T> whenComplete(action()) { | 274 Future<T> whenComplete(action()) { |
| 184 _Future result = new _Future<T>._whenComplete(action); | 275 return new _WhenFuture<T>(action).._subscribeTo(this); |
| 185 _addListener(result); | |
| 186 return result; | |
| 187 } | 276 } |
| 188 | 277 |
| 189 Stream<T> asStream() => new Stream.fromFuture(this); | 278 Stream<T> asStream() => new Stream.fromFuture(this); |
| 190 | 279 |
| 191 void _markPendingCompletion() { | 280 bool _inSameErrorZone(_Zone otherZone) { |
| 192 if (!_mayComplete) throw new StateError("Future already completed"); | 281 return _zone.inSameErrorZone(otherZone); |
| 193 _state = _PENDING_COMPLETE; | |
| 194 } | |
| 195 | |
| 196 void _clearPendingCompletion() { | |
| 197 assert(_state == _PENDING_COMPLETE); | |
| 198 _state = _INCOMPLETE; | |
| 199 } | |
| 200 | |
| 201 T get _value { | |
| 202 assert(_isComplete && _hasValue); | |
| 203 return _resultOrListeners; | |
| 204 } | |
| 205 | |
| 206 Object get _error { | |
| 207 assert(_isComplete && _hasError); | |
| 208 return _resultOrListeners; | |
| 209 } | 282 } |
| 210 | 283 |
| 211 void _setValue(T value) { | 284 void _setValue(T value) { |
| 212 assert(!_isComplete); // But may have a completion pending. | 285 if (!_mayComplete) throw new StateError("Future already completed"); |
| 286 _setValueUnchecked(value); |
| 287 } |
| 288 |
| 289 void _setValueUnchecked(T value) { |
| 290 _FutureListener listeners = _isChained ? null : _removeListeners(); |
| 213 _state = _VALUE; | 291 _state = _VALUE; |
| 214 _resultOrListeners = value; | 292 _resultOrListeners = value; |
| 293 while (listeners != null) { |
| 294 _FutureListener listener = listeners; |
| 295 listeners = listener._nextListener; |
| 296 listener._nextListener = null; |
| 297 listener._sendValue(value); |
| 298 } |
| 215 } | 299 } |
| 216 | 300 |
| 217 void _setError(Object error) { | 301 void _setError(Object error) { |
| 218 assert(!_isComplete); // But may have a completion pending. | 302 if (!_mayComplete) throw new StateError("Future already completed"); |
| 303 _setErrorUnchecked(error); |
| 304 } |
| 305 |
| 306 void _setErrorUnchecked(Object error) { |
| 307 _FutureListener listeners; |
| 308 bool hasListeners; |
| 309 if (_isChained) { |
| 310 listeners = null; |
| 311 hasListeners = (_state == _CHAINED); // and not _CHAINED_UNLISTENED. |
| 312 } else { |
| 313 listeners = _removeListeners(); |
| 314 hasListeners = (listeners != null); |
| 315 } |
| 316 |
| 219 _state = _ERROR; | 317 _state = _ERROR; |
| 220 _resultOrListeners = error; | 318 _resultOrListeners = error; |
| 319 |
| 320 if (!hasListeners) { |
| 321 // TODO(floitsch): Hook this into unhandled error handling. |
| 322 var error = _resultOrListeners; |
| 323 _zone.handleUncaughtError(error); |
| 324 return; |
| 325 } |
| 326 while (listeners != null) { |
| 327 _FutureListener listener = listeners; |
| 328 listeners = listener._nextListener; |
| 329 listener._nextListener = null; |
| 330 listener._sendError(error); |
| 331 } |
| 221 } | 332 } |
| 222 | 333 |
| 223 void _addListener(_Future listener) { | 334 void _asyncSetValue(T value) { |
| 335 if (!_mayComplete) throw new StateError("Future already completed"); |
| 336 _state = _PENDING_COMPLETE; |
| 337 runAsync(() { _setValueUnchecked(value); }); |
| 338 } |
| 339 |
| 340 void _asyncSetError(Object error) { |
| 341 if (!_mayComplete) throw new StateError("Future already completed"); |
| 342 _state = _PENDING_COMPLETE; |
| 343 runAsync(() { _setErrorUnchecked(error); }); |
| 344 } |
| 345 |
| 346 void _addListener(_FutureListener listener) { |
| 224 assert(listener._nextListener == null); | 347 assert(listener._nextListener == null); |
| 348 if (!listener._inSameErrorZone(_zone)) { |
| 349 listener = new _ErrorZoneBoundaryListener(listener); |
| 350 } |
| 351 if (_isChained) { |
| 352 _state = _CHAINED; // In case it was _CHAINED_UNLISTENED. |
| 353 _FutureImpl resultSource = _chainSource; |
| 354 resultSource._addListener(listener); |
| 355 return; |
| 356 } |
| 225 if (_isComplete) { | 357 if (_isComplete) { |
| 226 // Handle late listeners asynchronously. | 358 // Handle late listeners asynchronously. |
| 227 runAsync(() { | 359 runAsync(() { |
| 228 _propagateToListeners(this, listener); | 360 if (_hasValue) { |
| 361 T value = _resultOrListeners; |
| 362 listener._sendValue(value); |
| 363 } else { |
| 364 assert(_hasError); |
| 365 listener._sendError(_resultOrListeners); |
| 366 } |
| 229 }); | 367 }); |
| 230 } else { | 368 } else { |
| 369 assert(!_isComplete); |
| 231 listener._nextListener = _resultOrListeners; | 370 listener._nextListener = _resultOrListeners; |
| 232 _resultOrListeners = listener; | 371 _resultOrListeners = listener; |
| 233 } | 372 } |
| 234 } | 373 } |
| 235 | 374 |
| 236 _Future _removeListeners() { | 375 _FutureListener _removeListeners() { |
| 237 // Reverse listeners before returning them, so the resulting list is in | 376 // Reverse listeners before returning them, so the resulting list is in |
| 238 // subscription order. | 377 // subscription order. |
| 239 assert(!_isComplete); | 378 assert(!_isComplete); |
| 240 _Future current = _resultOrListeners; | 379 _FutureListener current = _resultOrListeners; |
| 241 _resultOrListeners = null; | 380 _resultOrListeners = null; |
| 242 _Future prev = null; | 381 _FutureListener prev = null; |
| 243 while (current != null) { | 382 while (current != null) { |
| 244 _Future next = current._nextListener; | 383 _FutureListener next = current._nextListener; |
| 245 current._nextListener = prev; | 384 current._nextListener = prev; |
| 246 prev = current; | 385 prev = current; |
| 247 current = next; | 386 current = next; |
| 248 } | 387 } |
| 249 return prev; | 388 return prev; |
| 250 } | 389 } |
| 251 | 390 |
| 252 static void _chainFutures(Future source, _Future target) { | 391 /** |
| 253 assert(!target._isComplete); | 392 * Make another [_FutureImpl] receive the result of this one. |
| 254 | 393 * |
| 255 // Mark the target as chained (and as such half-completed). | 394 * If this future is already complete, the [future] is notified |
| 256 target._isChained = true; | 395 * immediately. This function is only called during event resolution |
| 257 if (source is _Future) { | 396 * where it's acceptable to send an event. |
| 258 _Future internalFuture = source; | 397 */ |
| 259 if (internalFuture._isComplete) { | 398 void _chain(_FutureImpl future) { |
| 260 _propagateToListeners(internalFuture, target); | 399 if (!_isComplete) { |
| 400 future._chainFromFuture(this); |
| 401 } else if (_hasValue) { |
| 402 future._setValue(_resultOrListeners); |
| 403 } else { |
| 404 assert(_hasError); |
| 405 future._setError(_resultOrListeners); |
| 406 } |
| 407 } |
| 408 |
| 409 /** |
| 410 * Returns the future that this future is chained to. |
| 411 * |
| 412 * If that future is itself chained to something else, |
| 413 * get the [_chainSource] of that future instead, and make this |
| 414 * future chain directly to the earliest source. |
| 415 */ |
| 416 _FutureImpl get _chainSource { |
| 417 assert(_isChained); |
| 418 _FutureImpl future = _resultOrListeners; |
| 419 if (future._isChained) { |
| 420 future = _resultOrListeners = future._chainSource; |
| 421 } |
| 422 return future; |
| 423 } |
| 424 |
| 425 /** |
| 426 * Make this incomplete future end up with the same result as [resultSource]. |
| 427 * |
| 428 * This is done by moving all listeners to [resultSource] and forwarding all |
| 429 * future [_addListener] calls to [resultSource] directly. |
| 430 */ |
| 431 void _chainFromFuture(_FutureImpl resultSource) { |
| 432 assert(!_isComplete); |
| 433 assert(!_isChained); |
| 434 if (resultSource._isChained) { |
| 435 resultSource = resultSource._chainSource; |
| 436 } |
| 437 assert(!resultSource._isChained); |
| 438 if (identical(this, resultSource)) { |
| 439 // The only unchained future in a future dependency tree (as defined |
| 440 // by the chain-relations) is the "root" that every other future depends |
| 441 // on. The future we are adding is unchained, so if it is already in the |
| 442 // tree, it must be the root, so that's the only one we need to check |
| 443 // against to detect a cycle. |
| 444 _setError(new StateError("Cyclic future dependency.")); |
| 445 return; |
| 446 } |
| 447 _FutureListener cursor = _removeListeners(); |
| 448 bool hadListeners = cursor != null; |
| 449 while (cursor != null) { |
| 450 _FutureListener listener = cursor; |
| 451 cursor = cursor._nextListener; |
| 452 listener._nextListener = null; |
| 453 resultSource._addListener(listener); |
| 454 } |
| 455 // Listen with this future as well, so that when the other future completes, |
| 456 // this future will be completed as well. |
| 457 resultSource._addListener(this._asListener()); |
| 458 _resultOrListeners = resultSource; |
| 459 _state = hadListeners ? _CHAINED : _CHAINED_UNLISTENED; |
| 460 } |
| 461 |
| 462 /** |
| 463 * Helper function to handle the result of transforming an incoming event. |
| 464 * |
| 465 * If the result is itself a [Future], this future is linked to that |
| 466 * future's output. If not, this future is completed with the result. |
| 467 */ |
| 468 void _setOrChainValue(var result) { |
| 469 assert(!_isChained); |
| 470 assert(!_isComplete); |
| 471 if (result is Future) { |
| 472 // Result should be a Future<T>. |
| 473 if (result is _FutureImpl) { |
| 474 _FutureImpl chainFuture = result; |
| 475 chainFuture._chain(this); |
| 476 return; |
| 261 } else { | 477 } else { |
| 262 internalFuture._addListener(target); | 478 Future future = result; |
| 479 future.then(_setValue, |
| 480 onError: _setError); |
| 481 return; |
| 263 } | 482 } |
| 264 } else { | 483 } else { |
| 265 source.then((value) { | 484 // Result must be of type T. |
| 266 // Clear the is-chained bit, so that we can use the standard | 485 _setValue(result); |
| 267 // _complete method. | 486 } |
| 268 target._isChained = false; | 487 } |
| 269 target._complete(value); | 488 |
| 270 }, | 489 _FutureListener _asListener() => new _FutureListener.wrap(this); |
| 271 onError: (error) { | 490 } |
| 272 // Clear the is-chained bit, so that we can use the standard | 491 |
| 273 // _completeError method. | 492 /** |
| 274 target._isChained = false; | 493 * Transforming future base class. |
| 275 target._completeError(error); | 494 * |
| 276 }); | 495 * A transforming future is itself a future and a future listener. |
| 277 } | 496 * Subclasses override [_sendValue]/[_sendError] to intercept |
| 278 } | 497 * the results of a previous future. |
| 279 | 498 */ |
| 280 void _complete(value) { | 499 abstract class _TransformFuture<S, T> extends _FutureImpl<T> |
| 281 assert(_onValueCallback == null && | 500 implements _FutureListener<S> { |
| 282 _onErrorCallback == null && | 501 // _FutureListener implementation. |
| 283 _whenCompleteActionCallback == null && | 502 _FutureListener _nextListener; |
| 284 _errorTestCallback == null); | 503 |
| 285 if (!_mayComplete) throw new StateError("Future already completed"); | 504 _TransformFuture() { |
| 286 if (value is Future) { | 505 _zone.expectCallback(); |
| 287 _chainFutures(value, this); | 506 } |
| 288 return; | 507 |
| 289 } | 508 void _sendValue(S value) { |
| 290 _Future listeners = _removeListeners(); | 509 _zone.executeCallback(() => _zonedSendValue(value)); |
| 510 } |
| 511 |
| 512 void _sendError(error) { |
| 513 _zone.executeCallback(() => _zonedSendError(error)); |
| 514 } |
| 515 |
| 516 void _subscribeTo(_FutureImpl future) { |
| 517 future._addListener(this); |
| 518 } |
| 519 |
| 520 void _zonedSendValue(S value); |
| 521 void _zonedSendError(error); |
| 522 } |
| 523 |
| 524 /** The onValue and onError handlers return either a value or a future */ |
| 525 typedef dynamic _FutureOnValue<T>(T value); |
| 526 typedef dynamic _FutureOnError(error); |
| 527 /** Test used by [Future.catchError] to handle skip some errors. */ |
| 528 typedef bool _FutureErrorTest(var error); |
| 529 /** Used by [WhenFuture]. */ |
| 530 typedef _FutureAction(); |
| 531 |
| 532 /** Future returned by [Future.then] with no [:onError:] parameter. */ |
| 533 class _ThenFuture<S, T> extends _TransformFuture<S, T> { |
| 534 // TODO(ahe): Restore type when feature is implemented in dart2js |
| 535 // checked mode. |
| 536 final /* _FutureOnValue<S> */ _onValue; |
| 537 |
| 538 _ThenFuture(this._onValue); |
| 539 |
| 540 _zonedSendValue(S value) { |
| 541 assert(_onValue != null); |
| 542 var result; |
| 543 try { |
| 544 result = _onValue(value); |
| 545 } catch (e, s) { |
| 546 _setError(_asyncError(e, s)); |
| 547 return; |
| 548 } |
| 549 _setOrChainValue(result); |
| 550 } |
| 551 |
| 552 void _zonedSendError(error) { |
| 553 _setError(error); |
| 554 } |
| 555 } |
| 556 |
| 557 /** Future returned by [Future.catchError]. */ |
| 558 class _CatchErrorFuture<T> extends _TransformFuture<T,T> { |
| 559 final _FutureErrorTest _test; |
| 560 final _FutureOnError _onError; |
| 561 |
| 562 _CatchErrorFuture(this._onError, this._test); |
| 563 |
| 564 _zonedSendValue(T value) { |
| 291 _setValue(value); | 565 _setValue(value); |
| 292 _propagateToListeners(this, listeners); | 566 } |
| 293 } | 567 |
| 294 | 568 _zonedSendError(error) { |
| 295 void _completeError(error, [StackTrace stackTrace]) { | 569 assert(_onError != null); |
| 296 assert(_onValueCallback == null); | 570 // if _test is supplied, check if it returns true, otherwise just |
| 297 assert(_onErrorCallback == null); | 571 // forward the error unmodified. |
| 298 assert(_whenCompleteActionCallback == null); | 572 if (_test != null) { |
| 299 assert(_errorTestCallback == null); | 573 bool matchesTest; |
| 300 // _isComplete does not trigger for pending completions. | 574 try { |
| 301 if (!_mayComplete) throw new StateError("Future already completed"); | 575 matchesTest = _test(error); |
| 302 if (stackTrace != null) { | 576 } catch (e, s) { |
| 303 // Force the stack trace onto the error, even if it already had one. | 577 _setError(_asyncError(e, s)); |
| 304 _attachStackTrace(error, stackTrace); | 578 return; |
| 305 } | 579 } |
| 306 | 580 if (!matchesTest) { |
| 307 _Future listeners = _isChained ? null : _removeListeners(); | 581 _setError(error); |
| 582 return; |
| 583 } |
| 584 } |
| 585 // Act on the error, and use the result as this future's result. |
| 586 var result; |
| 587 try { |
| 588 result = _onError(error); |
| 589 } catch (e, s) { |
| 590 _setError(_asyncError(e, s)); |
| 591 return; |
| 592 } |
| 593 _setOrChainValue(result); |
| 594 } |
| 595 } |
| 596 |
| 597 /** Future returned by [Future.then] with an [:onError:] parameter. */ |
| 598 class _SubscribeFuture<S, T> extends _ThenFuture<S, T> { |
| 599 final _FutureOnError _onError; |
| 600 |
| 601 _SubscribeFuture(onValue(S value), this._onError) : super(onValue); |
| 602 |
| 603 // The _sendValue method is inherited from ThenFuture. |
| 604 |
| 605 void _zonedSendError(error) { |
| 606 assert(_onError != null); |
| 607 var result; |
| 608 try { |
| 609 result = _onError(error); |
| 610 } catch (e, s) { |
| 611 _setError(_asyncError(e, s)); |
| 612 return; |
| 613 } |
| 614 _setOrChainValue(result); |
| 615 } |
| 616 } |
| 617 |
| 618 /** Future returned by [Future.whenComplete]. */ |
| 619 class _WhenFuture<T> extends _TransformFuture<T, T> { |
| 620 final _FutureAction _action; |
| 621 |
| 622 _WhenFuture(this._action); |
| 623 |
| 624 void _zonedSendValue(T value) { |
| 625 try { |
| 626 var result = _action(); |
| 627 if (result is Future) { |
| 628 Future resultFuture = result; |
| 629 resultFuture.then((_) { |
| 630 _setValue(value); |
| 631 }, onError: _setError); |
| 632 return; |
| 633 } |
| 634 } catch (e, s) { |
| 635 _setError(_asyncError(e, s)); |
| 636 return; |
| 637 } |
| 638 _setValue(value); |
| 639 } |
| 640 |
| 641 void _zonedSendError(error) { |
| 642 try { |
| 643 var result = _action(); |
| 644 if (result is Future) { |
| 645 Future resultFuture = result; |
| 646 // TODO(lrn): Find a way to combine [error] into [e]. |
| 647 resultFuture.then((_) { |
| 648 _setError(error); |
| 649 }, onError: _setError); |
| 650 return; |
| 651 } |
| 652 } catch (e, s) { |
| 653 error = _asyncError(e, s); |
| 654 } |
| 308 _setError(error); | 655 _setError(error); |
| 309 _propagateToListeners(this, listeners); | 656 } |
| 310 } | 657 } |
| 311 | |
| 312 void _asyncComplete(value) { | |
| 313 assert(_onValueCallback == null); | |
| 314 assert(_onErrorCallback == null); | |
| 315 assert(_whenCompleteActionCallback == null); | |
| 316 assert(_errorTestCallback == null); | |
| 317 if (!_mayComplete) throw new StateError("Future already completed"); | |
| 318 // Two corner cases if the value is a future: | |
| 319 // 1. the future is already completed and an error. | |
| 320 // 2. the future is not yet completed but might become an error. | |
| 321 // The first case means that we must not immediately complete the Future, | |
| 322 // as our code would immediately start propagating the error without | |
| 323 // giving the time to install error-handlers. | |
| 324 // However the second case requires us to deal with the value immediately. | |
| 325 // Otherwise the value could complete with an error and report an | |
| 326 // unhandled error, even though we know we are already going to listen to | |
| 327 // it. | |
| 328 if (value is Future && | |
| 329 (value is! _Future || !(value as _Future)._isComplete)) { | |
| 330 // Case 2 from above. We need to register. | |
| 331 // Note that we are still completing asynchronously: either we register | |
| 332 // through .then (in which case the completing is asynchronous), or we | |
| 333 // have a _Future which isn't complete yet. | |
| 334 _complete(value); | |
| 335 return; | |
| 336 } | |
| 337 | |
| 338 _markPendingCompletion(); | |
| 339 runAsync(() { | |
| 340 _clearPendingCompletion(); | |
| 341 _complete(value); | |
| 342 }); | |
| 343 } | |
| 344 | |
| 345 void _asyncCompleteError(error, [StackTrace stackTrace]) { | |
| 346 assert(_onValueCallback == null); | |
| 347 assert(_onErrorCallback == null); | |
| 348 assert(_whenCompleteActionCallback == null); | |
| 349 assert(_errorTestCallback == null); | |
| 350 if (!_mayComplete) throw new StateError("Future already completed"); | |
| 351 _markPendingCompletion(); | |
| 352 runAsync(() { | |
| 353 _clearPendingCompletion(); | |
| 354 _completeError(error, stackTrace); | |
| 355 }); | |
| 356 } | |
| 357 | |
| 358 /** | |
| 359 * Propagates the value/error of [source] to its [listeners]. | |
| 360 * | |
| 361 * Unlinks all listeners and propagates the source to each listener | |
| 362 * separately. | |
| 363 */ | |
| 364 static void _propagateMultipleListeners(_Future source, _Future listeners) { | |
| 365 assert(listeners != null); | |
| 366 assert(listeners._nextListener != null); | |
| 367 do { | |
| 368 _Future listener = listeners; | |
| 369 listeners = listener._nextListener; | |
| 370 listener._nextListener = null; | |
| 371 _propagateToListeners(source, listener); | |
| 372 } while (listeners != null); | |
| 373 } | |
| 374 | |
| 375 /** | |
| 376 * Propagates the value/error of [source] to its [listeners], executing the | |
| 377 * listeners' callbacks. | |
| 378 * | |
| 379 * If [runCallback] is true (which should be the default) it executes | |
| 380 * the registered action of listeners. If it is `false` then the callback is | |
| 381 * skipped. This is used to complete futures with chained futures. | |
| 382 */ | |
| 383 static void _propagateToListeners(_Future source, _Future listeners) { | |
| 384 while (true) { | |
| 385 if (!source._isComplete) return; // Chained future. | |
| 386 bool hasError = source._hasError; | |
| 387 if (hasError && listeners == null) { | |
| 388 source._zone.handleUncaughtError(source._error); | |
| 389 return; | |
| 390 } | |
| 391 if (listeners == null) return; | |
| 392 _Future listener = listeners; | |
| 393 if (listener._nextListener != null) { | |
| 394 // Usually futures only have one listener. If they have several, we | |
| 395 // handle them specially. | |
| 396 _propagateMultipleListeners(source, listeners); | |
| 397 return; | |
| 398 } | |
| 399 if (hasError && !source._zone.inSameErrorZone(listener._zone)) { | |
| 400 // Don't cross zone boundaries with errors. | |
| 401 source._zone.handleUncaughtError(source._error); | |
| 402 return; | |
| 403 } | |
| 404 if (!identical(_Zone.current, listener._zone)) { | |
| 405 // Run the propagation in the listener's zone to avoid | |
| 406 // zone transitions. The idea is that many chained futures will | |
| 407 // be in the same zone. | |
| 408 listener._zone.executePeriodicCallback(() { | |
| 409 _propagateToListeners(source, listener); | |
| 410 }); | |
| 411 return; | |
| 412 } | |
| 413 | |
| 414 // Do the actual propagation. | |
| 415 // TODO(floitsch): Do we need to go through the zone even if we | |
| 416 // don't have a callback to execute? | |
| 417 bool listenerHasValue; | |
| 418 var listenerValueOrError; | |
| 419 // Set to true if a whenComplete needs to wait for a future. | |
| 420 // The whenComplete action will resume the propagation by itself. | |
| 421 bool isPropagationAborted = false; | |
| 422 // Even though we are already in the right zone (due to the optimization | |
| 423 // above), we still need to go through the zone. The overhead of | |
| 424 // executeCallback is however smaller when it is already in the correct | |
| 425 // zone. | |
| 426 // TODO(floitsch): only run callbacks in the zone, not the whole | |
| 427 // handling code. | |
| 428 listener._zone.executeCallback(() { | |
| 429 // TODO(floitsch): mark the listener as pending completion. Currently | |
| 430 // we can't do this, since the markPendingCompletion verifies that | |
| 431 // the future is not already marked (or chained). | |
| 432 try { | |
| 433 if (!hasError) { | |
| 434 var value = source._value; | |
| 435 if (listener._onValue != null) { | |
| 436 listenerValueOrError = listener._onValue(value); | |
| 437 listenerHasValue = true; | |
| 438 } else { | |
| 439 // Copy over the value from the source. | |
| 440 listenerValueOrError = value; | |
| 441 listenerHasValue = true; | |
| 442 } | |
| 443 } else { | |
| 444 Object error = source._error; | |
| 445 _FutureErrorTest test = listener._errorTest; | |
| 446 bool matchesTest = true; | |
| 447 if (test != null) { | |
| 448 matchesTest = test(error); | |
| 449 } | |
| 450 if (matchesTest && listener._onError != null) { | |
| 451 listenerValueOrError = listener._onError(error); | |
| 452 listenerHasValue = true; | |
| 453 } else { | |
| 454 // Copy over the error from the source. | |
| 455 listenerValueOrError = error; | |
| 456 listenerHasValue = false; | |
| 457 } | |
| 458 } | |
| 459 | |
| 460 if (listener._whenCompleteAction != null) { | |
| 461 var completeResult = listener._whenCompleteAction(); | |
| 462 if (completeResult is Future) { | |
| 463 listener._isChained = true; | |
| 464 completeResult.then((ignored) { | |
| 465 // Try again, but this time don't run the whenComplete callback. | |
| 466 _propagateToListeners(source, listener); | |
| 467 }, onError: (error) { | |
| 468 // When there is an error, we have to make the error the new | |
| 469 // result of the current listener. | |
| 470 if (completeResult is! _Future) { | |
| 471 // This should be a rare case. | |
| 472 completeResult = new _Future(); | |
| 473 completeResult._setError(error); | |
| 474 } | |
| 475 _propagateToListeners(completeResult, listener); | |
| 476 }); | |
| 477 isPropagationAborted = true; | |
| 478 // We will reenter the listener's zone. | |
| 479 listener._zone.expectCallback(); | |
| 480 } | |
| 481 } | |
| 482 } catch (e, s) { | |
| 483 // Set the exception as error. | |
| 484 listenerValueOrError = _asyncError(e, s); | |
| 485 listenerHasValue = false; | |
| 486 } | |
| 487 if (listenerHasValue && listenerValueOrError is Future) { | |
| 488 // We are going to reenter the zone to finish what we started. | |
| 489 listener._zone.expectCallback(); | |
| 490 } | |
| 491 }); | |
| 492 if (isPropagationAborted) return; | |
| 493 // If the listener's value is a future we need to chain it. | |
| 494 if (listenerHasValue && listenerValueOrError is Future) { | |
| 495 Future chainSource = listenerValueOrError; | |
| 496 // Shortcut if the chain-source is already completed. Just continue the | |
| 497 // loop. | |
| 498 if (chainSource is _Future && (chainSource as _Future)._isComplete) { | |
| 499 // propagate the value (simulating a tail call). | |
| 500 listener._isChained = true; | |
| 501 source = chainSource; | |
| 502 listeners = listener; | |
| 503 continue; | |
| 504 } | |
| 505 _chainFutures(chainSource, listener); | |
| 506 return; | |
| 507 } | |
| 508 | |
| 509 if (listenerHasValue) { | |
| 510 listeners = listener._removeListeners(); | |
| 511 listener._setValue(listenerValueOrError); | |
| 512 } else { | |
| 513 listeners = listener._removeListeners(); | |
| 514 listener._setError(listenerValueOrError); | |
| 515 } | |
| 516 // Prepare for next round. | |
| 517 source = listener; | |
| 518 } | |
| 519 } | |
| 520 } | |
| OLD | NEW |