Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(416)

Side by Side Diff: sdk/lib/async/future_impl.dart

Issue 23967011: Revert "Rewrite Futures." (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Created 7 years, 3 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch | Annotate | Revision Log
« no previous file with comments | « sdk/lib/async/future.dart ('k') | sdk/lib/async/stream.dart » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
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 }
OLDNEW
« no previous file with comments | « sdk/lib/async/future.dart ('k') | sdk/lib/async/stream.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698