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

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

Issue 637893002: Use a _FutureListener separate from the _Future to hold listeners. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Don't check that source is complete, it always is. Created 6 years, 2 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 | « no previous file | sdk/lib/async/schedule_microtask.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 /** The onValue and onError handlers return either a value or a future */
8 typedef dynamic _FutureOnValue<T>(T value); 8 typedef dynamic _FutureOnValue<T>(T value);
9 /** Test used by [Future.catchError] to handle skip some errors. */ 9 /** Test used by [Future.catchError] to handle skip some errors. */
10 typedef bool _FutureErrorTest(var error); 10 typedef bool _FutureErrorTest(var error);
(...skipping 39 matching lines...) Expand 10 before | Expand all | Expand 10 after
50 void complete([value]) { 50 void complete([value]) {
51 if (!future._mayComplete) throw new StateError("Future already completed"); 51 if (!future._mayComplete) throw new StateError("Future already completed");
52 future._complete(value); 52 future._complete(value);
53 } 53 }
54 54
55 void _completeError(Object error, StackTrace stackTrace) { 55 void _completeError(Object error, StackTrace stackTrace) {
56 future._completeError(error, stackTrace); 56 future._completeError(error, stackTrace);
57 } 57 }
58 } 58 }
59 59
60 class _FutureListener {
61 static const int MASK_VALUE = 1;
62 static const int MASK_ERROR = 2;
63 static const int MASK_TEST_ERROR = 4;
64 static const int MASK_WHENCOMPLETE = 8;
65 static const int STATE_CHAIN = 0;
66 static const int STATE_THEN = MASK_VALUE;
67 static const int STATE_THEN_ONERROR = MASK_VALUE | MASK_ERROR;
68 static const int STATE_CATCHERROR = MASK_ERROR;
69 static const int STATE_CATCHERROR_TEST = MASK_ERROR | MASK_TEST_ERROR;
70 static const int STATE_WHENCOMPLETE = MASK_WHENCOMPLETE;
71 // Listeners on the same future are linked through this link.
72 _FutureListener _nextListener = null;
73 // The future to complete when this listener is activated.
74 final _Future result;
75 // Which fields means what.
76 final int state;
77 // Used for then/whenDone callback and error test
78 final Function callback;
79 // Used for error callbacks.
80 final Function errorCallback;
81
82 _FutureListener.then(this.result,
83 _FutureOnValue onValue, Function errorCallback)
Søren Gjesse 2014/10/09 06:55:51 No typedef for errorCallback due to stack trace or
Lasse Reichstein Nielsen 2014/10/09 07:53:21 Yes. It can be an either one- or two-argument func
84 : callback = onValue,
85 errorCallback = errorCallback,
86 state = (errorCallback == null) ? STATE_THEN : STATE_THEN_ONERROR;
87
88 _FutureListener.catchError(this.result,
89 this.errorCallback, _FutureErrorTest test)
90 : callback = test,
91 state = (test == null) ? STATE_CATCHERROR : STATE_CATCHERROR_TEST;
92
93 _FutureListener.whenComplete(this.result, _FutureAction onComplete)
94 : callback = onComplete,
95 errorCallback = null,
96 state = STATE_WHENCOMPLETE;
97
98 _FutureListener.chain(this.result)
99 : callback = null,
100 errorCallback = null,
101 state = STATE_CHAIN;
102
103 Zone get _zone => result._zone;
104
105 bool get handlesValue => (state & MASK_VALUE != 0);
106 bool get handlesError => (state & MASK_ERROR != 0);
107 bool get hasErrorTest => (state == STATE_CATCHERROR_TEST);
108 bool get handlesComplete => (state == STATE_WHENCOMPLETE);
109
110 _FutureOnValue get _onValue {
111 assert(handlesValue);
112 return callback;
113 }
114 Function get _onError => errorCallback;
115 _FutureErrorTest get _errorTest {
116 assert(hasErrorTest);
117 return callback;
118 }
119 _FutureAction get _whenCompleteAction {
120 assert(handlesComplete);
121 return callback;
122 }
123 }
124
60 class _Future<T> implements Future<T> { 125 class _Future<T> implements Future<T> {
61 // State of the future. The state determines the interpretation of the
62 // [resultOrListeners] field.
63 // TODO(lrn): rename field since it can also contain a chained future.
64
65 /// Initial state, waiting for a result. In this state, the 126 /// Initial state, waiting for a result. In this state, the
66 /// [resultOrListeners] field holds a single-linked list of 127 /// [resultOrListeners] field holds a single-linked list of
67 /// [FutureListener] listeners. 128 /// [_FutureListener] listeners.
68 static const int _INCOMPLETE = 0; 129 static const int _INCOMPLETE = 0;
69 /// Pending completion. Set when completed using [_asyncComplete] or 130 /// Pending completion. Set when completed using [_asyncComplete] or
70 /// [_asyncCompleteError]. It is an error to try to complete it again. 131 /// [_asyncCompleteError]. It is an error to try to complete it again.
132 /// [resultOrListeners] holds listeners.
71 static const int _PENDING_COMPLETE = 1; 133 static const int _PENDING_COMPLETE = 1;
72 /// The future has been chained to another future. The result of that 134 /// The future has been chained to another future. The result of that
73 /// other future becomes the result of this future as well. 135 /// other future becomes the result of this future as well.
74 /// In this state, no callback should be executed anymore. 136 /// In this state, no callback should be executed anymore.
75 // TODO(floitsch): we don't really need a special "_CHAINED" state. We could 137 // TODO(floitsch): we don't really need a special "_CHAINED" state. We could
76 // just use the PENDING_COMPLETE state instead. 138 // just use the PENDING_COMPLETE state instead.
77 static const int _CHAINED = 2; 139 static const int _CHAINED = 2;
78 /// The future has been completed with a value result. 140 /// The future has been completed with a value result.
79 static const int _VALUE = 4; 141 static const int _VALUE = 4;
80 /// The future has been completed with an error result. 142 /// The future has been completed with an error result.
81 static const int _ERROR = 8; 143 static const int _ERROR = 8;
82 144
83 /** Whether the future is complete, and as what. */ 145 /** Whether the future is complete, and as what. */
84 int _state = _INCOMPLETE; 146 int _state = _INCOMPLETE;
85 147
86 final Zone _zone; 148 /**
87 149 * Zone that the future was completed from.
88 bool get _mayComplete => _state == _INCOMPLETE; 150 * This is the zone that an error result belongs to.
89 bool get _isChained => _state == _CHAINED; 151 *
90 bool get _isComplete => _state >= _VALUE; 152 * Until the future is completed, the field may hold the zone that
91 bool get _hasValue => _state == _VALUE; 153 * listener callbacks used to create this future should be run in.
92 bool get _hasError => _state == _ERROR; 154 */
93 155 final Zone _zone = Zone.current;
94 set _isChained(bool value) {
95 if (value) {
96 assert(!_isComplete);
97 _state = _CHAINED;
98 } else {
99 assert(_isChained);
100 _state = _INCOMPLETE;
101 }
102 }
103 156
104 /** 157 /**
105 * Either the result, a list of listeners or another future. 158 * Either the result, a list of listeners or another future.
106 * 159 *
107 * The result of the future is either a value or an error. 160 * The result of the future is either a value or an error.
108 * A result is only stored when the future has completed. 161 * A result is only stored when the future has completed.
109 * 162 *
110 * The listeners is an internally linked list of [_FutureListener]s. 163 * The listeners is an internally linked list of [_FutureListener]s.
Søren Gjesse 2014/10/09 06:55:51 So the _FutureListener class have existed before?
Lasse Reichstein Nielsen 2014/10/09 07:53:22 Yes, and the documentation wasn't updated when it
111 * Listeners are only remembered while the future is not yet complete, 164 * Listeners are only remembered while the future is not yet complete,
112 * and it is not chained to another future. 165 * and it is not chained to another future.
113 * 166 *
114 * The future is another future that his future is chained to. This future 167 * The future is another future that his future is chained to. This future
115 * is waiting for the other future to complete, and when it does, this future 168 * is waiting for the other future to complete, and when it does, this future
116 * will complete with the same result. 169 * will complete with the same result.
117 * All listeners are forwarded to the other future. 170 * All listeners are forwarded to the other future.
118 * 171 *
119 * The cases are disjoint (incomplete and unchained, incomplete and 172 * The cases are disjoint (incomplete and unchained, incomplete and
120 * chained, or completed with value or error), so the field only needs to hold 173 * chained, or completed with value or error), so the field only needs to hold
121 * one value at a time. 174 * one value at a time.
Søren Gjesse 2014/10/09 06:55:52 Please update this with the mapping to the state v
Lasse Reichstein Nielsen 2014/10/09 07:53:22 Done.
122 */ 175 */
123 var _resultOrListeners; 176 var _resultOrListeners;
124 177
125 /** 178 _Future();
126 * A [_Future] implements a linked list. If a future has more than one
127 * listener the [_nextListener] field of the first listener points to the
128 * remaining listeners.
129 */
130 // TODO(floitsch): since single listeners are the common case we should
131 // use a bit to indicate that the _resultOrListeners contains a container.
132 _Future _nextListener;
133
134 // TODO(floitsch): we only need two closure fields to store the callbacks.
135 // If we store the type of a closure in the state field (where there are
136 // still bits left), we can just store two closures instead of using 4
137 // fields of which 2 are always null.
138 _FutureOnValue _onValueCallback;
139 _FutureErrorTest _errorTestCallback;
140 Function _onErrorCallback;
141 _FutureAction _whenCompleteActionCallback;
142
143 _FutureOnValue get _onValue => _isChained ? null : _onValueCallback;
144 _FutureErrorTest get _errorTest => _isChained ? null : _errorTestCallback;
145 Function get _onError => _isChained ? null : _onErrorCallback;
146 _FutureAction get _whenCompleteAction
147 => _isChained ? null : _whenCompleteActionCallback;
148
149 _Future()
150 : _zone = Zone.current,
151 _onValueCallback = null, _errorTestCallback = null,
152 _onErrorCallback = null, _whenCompleteActionCallback = null;
153 179
154 /// Valid types for value: `T` or `Future<T>`. 180 /// Valid types for value: `T` or `Future<T>`.
155 _Future.immediate(value) 181 _Future.immediate(value) {
156 : _zone = Zone.current,
157 _onValueCallback = null, _errorTestCallback = null,
158 _onErrorCallback = null, _whenCompleteActionCallback = null {
159 _asyncComplete(value); 182 _asyncComplete(value);
160 } 183 }
161 184
162 _Future.immediateError(var error, [StackTrace stackTrace]) 185 _Future.immediateError(var error, [StackTrace stackTrace]) {
163 : _zone = Zone.current,
164 _onValueCallback = null, _errorTestCallback = null,
165 _onErrorCallback = null, _whenCompleteActionCallback = null {
166 _asyncCompleteError(error, stackTrace); 186 _asyncCompleteError(error, stackTrace);
167 } 187 }
168 188
169 _Future._then(onValueCallback(value), Function onErrorCallback) 189 bool get _mayComplete => _state == _INCOMPLETE;
170 : _zone = Zone.current, 190 bool get _isChained => _state == _CHAINED;
171 _onValueCallback = Zone.current.registerUnaryCallback(onValueCallback), 191 bool get _isComplete => _state >= _VALUE;
172 _onErrorCallback = _registerErrorHandler(onErrorCallback, Zone.current), 192 bool get _hasValue => _state == _VALUE;
173 _errorTestCallback = null, 193 bool get _hasError => _state == _ERROR;
174 _whenCompleteActionCallback = null;
175 194
176 _Future._catchError(Function onErrorCallback, bool errorTestCallback(e)) 195 set _isChained(bool value) {
177 : _zone = Zone.current, 196 if (value) {
178 _onErrorCallback = _registerErrorHandler(onErrorCallback, Zone.current), 197 assert(!_isComplete);
179 _errorTestCallback = 198 _state = _CHAINED;
180 Zone.current.registerUnaryCallback(errorTestCallback), 199 } else {
181 _onValueCallback = null, 200 assert(_isChained);
182 _whenCompleteActionCallback = null; 201 _state = _INCOMPLETE;
183 202 }
184 _Future._whenComplete(whenCompleteActionCallback()) 203 }
185 : _zone = Zone.current,
186 _whenCompleteActionCallback =
187 Zone.current.registerCallback(whenCompleteActionCallback),
188 _onValueCallback = null,
189 _errorTestCallback = null,
190 _onErrorCallback = null;
191 204
192 Future then(f(T value), { Function onError }) { 205 Future then(f(T value), { Function onError }) {
193 _Future result; 206 _Future result;
194 result = new _Future._then(f, onError); 207 result = new _Future();
Søren Gjesse 2014/10/09 06:55:51 Combine the declaration with first assignment.
Lasse Reichstein Nielsen 2014/10/09 07:53:22 Done.
195 _addListener(result); 208 if (!identical(result._zone, _ROOT_ZONE)) {
209 f = result._zone.registerUnaryCallback(f);
210 if (onError != null) {
211 onError = _registerErrorHandler(onError, result._zone);
212 }
213 }
214 _addListener(new _FutureListener.then(result, f, onError));
196 return result; 215 return result;
197 } 216 }
198 217
199 Future catchError(Function onError, { bool test(error) }) { 218 Future catchError(Function onError, { bool test(error) }) {
200 _Future result = new _Future._catchError(onError, test); 219 _Future result = new _Future();
201 _addListener(result); 220 if (!identical(result._zone, _ROOT_ZONE)) {
221 onError = _registerErrorHandler(onError, result._zone);
222 if (test != null) test = result._zone.registerUnaryCallback(test);
223 }
224 _addListener(new _FutureListener.catchError(result, onError, test));
202 return result; 225 return result;
203 } 226 }
204 227
205 Future<T> whenComplete(action()) { 228 Future<T> whenComplete(action()) {
206 _Future result = new _Future<T>._whenComplete(action); 229 _Future result = new _Future<T>();
207 _addListener(result); 230 if (!identical(result._zone, _ROOT_ZONE)) {
231 action = result._zone.registerCallback(action);
232 }
233 _addListener(new _FutureListener.whenComplete(result, action));
208 return result; 234 return result;
209 } 235 }
210 236
211 Stream<T> asStream() => new Stream.fromFuture(this); 237 Stream<T> asStream() => new Stream.fromFuture(this);
212 238
213 void _markPendingCompletion() { 239 void _markPendingCompletion() {
214 if (!_mayComplete) throw new StateError("Future already completed"); 240 if (!_mayComplete) throw new StateError("Future already completed");
215 _state = _PENDING_COMPLETE; 241 _state = _PENDING_COMPLETE;
216 } 242 }
217 243
218 T get _value { 244 T get _value {
219 assert(_isComplete && _hasValue); 245 assert(_isComplete && _hasValue);
220 return _resultOrListeners; 246 return _resultOrListeners;
221 } 247 }
222 248
223 AsyncError get _error { 249 AsyncError get _error {
224 assert(_isComplete && _hasError); 250 assert(_isComplete && _hasError);
225 return _resultOrListeners; 251 return _resultOrListeners;
226 } 252 }
227 253
228 void _setValue(T value) { 254 void _setValue(T value) {
229 assert(!_isComplete); // But may have a completion pending. 255 assert(!_isComplete); // But may have a completion pending.
230 _state = _VALUE; 256 _state = _VALUE;
231 _resultOrListeners = value; 257 _resultOrListeners = value;
232 } 258 }
233 259
234 void _setError(Object error, StackTrace stackTrace) { 260 void _setErrorObject(AsyncError error) {
235 assert(!_isComplete); // But may have a completion pending. 261 assert(!_isComplete); // But may have a completion pending.
236 _state = _ERROR; 262 _state = _ERROR;
237 _resultOrListeners = new AsyncError(error, stackTrace); 263 _resultOrListeners = error;
238 } 264 }
239 265
240 void _addListener(_Future listener) { 266 void _setError(Object error, StackTrace stackTrace) {
267 _setErrorObject(new AsyncError(error, stackTrace));
268 }
269
270 void _addListener(_FutureListener listener) {
241 assert(listener._nextListener == null); 271 assert(listener._nextListener == null);
242 if (_isComplete) { 272 if (_isComplete) {
243 // Handle late listeners asynchronously. 273 // Handle late listeners asynchronously.
244 _zone.scheduleMicrotask(() { 274 _zone.scheduleMicrotask(() {
245 _propagateToListeners(this, listener); 275 _propagateToListeners(this, listener);
246 }); 276 });
247 } else { 277 } else {
248 listener._nextListener = _resultOrListeners; 278 listener._nextListener = _resultOrListeners;
249 _resultOrListeners = listener; 279 _resultOrListeners = listener;
250 } 280 }
251 } 281 }
252 282
253 _Future _removeListeners() { 283 _FutureListener _removeListeners() {
254 // Reverse listeners before returning them, so the resulting list is in 284 // Reverse listeners before returning them, so the resulting list is in
255 // subscription order. 285 // subscription order.
256 assert(!_isComplete); 286 assert(!_isComplete);
257 _Future current = _resultOrListeners; 287 _FutureListener current = _resultOrListeners;
258 _resultOrListeners = null; 288 _resultOrListeners = null;
259 _Future prev = null; 289 _FutureListener prev = null;
260 while (current != null) { 290 while (current != null) {
261 _Future next = current._nextListener; 291 _FutureListener next = current._nextListener;
262 current._nextListener = prev; 292 current._nextListener = prev;
263 prev = current; 293 prev = current;
264 current = next; 294 current = next;
265 } 295 }
266 return prev; 296 return prev;
267 } 297 }
268 298
269 // Take the value (when completed) of source and complete target with that 299 // Take the value (when completed) of source and complete target with that
270 // value (or error). This function can chain all Futures, but is slower 300 // value (or error). This function can chain all Futures, but is slower
271 // for _Future than _chainCoreFuture - Use _chainCoreFuture in that case. 301 // for _Future than _chainCoreFuture - Use _chainCoreFuture in that case.
(...skipping 18 matching lines...) Expand all
290 } 320 }
291 321
292 // Take the value (when completed) of source and complete target with that 322 // Take the value (when completed) of source and complete target with that
293 // value (or error). This function expects that source is a _Future. 323 // value (or error). This function expects that source is a _Future.
294 static void _chainCoreFuture(_Future source, _Future target) { 324 static void _chainCoreFuture(_Future source, _Future target) {
295 assert(!target._isComplete); 325 assert(!target._isComplete);
296 assert(source is _Future); 326 assert(source is _Future);
297 327
298 // Mark the target as chained (and as such half-completed). 328 // Mark the target as chained (and as such half-completed).
299 target._isChained = true; 329 target._isChained = true;
300 _Future internalFuture = source; 330 _FutureListener listener = new _FutureListener.chain(target);
301 if (internalFuture._isComplete) { 331 if (source._isComplete) {
302 _propagateToListeners(internalFuture, target); 332 _propagateToListeners(source, listener);
303 } else { 333 } else {
304 internalFuture._addListener(target); 334 source._addListener(listener);
305 } 335 }
306 } 336 }
307 337
308 void _complete(value) { 338 void _complete(value) {
309 assert(!_isComplete); 339 assert(!_isComplete);
310 assert(_onValue == null);
311 assert(_onError == null);
312 assert(_whenCompleteAction == null);
313 assert(_errorTest == null);
314
315 if (value is Future) { 340 if (value is Future) {
316 if (value is _Future) { 341 if (value is _Future) {
317 _chainCoreFuture(value, this); 342 _chainCoreFuture(value, this);
318 } else { 343 } else {
319 _chainForeignFuture(value, this); 344 _chainForeignFuture(value, this);
320 } 345 }
321 } else { 346 } else {
322 _Future listeners = _removeListeners(); 347 _FutureListener listeners = _removeListeners();
323 _setValue(value); 348 _setValue(value);
324 _propagateToListeners(this, listeners); 349 _propagateToListeners(this, listeners);
325 } 350 }
326 } 351 }
327 352
328 void _completeWithValue(value) { 353 void _completeWithValue(value) {
329 assert(!_isComplete); 354 assert(!_isComplete);
330 assert(_onValue == null);
331 assert(_onError == null);
332 assert(_whenCompleteAction == null);
333 assert(_errorTest == null);
334 assert(value is! Future); 355 assert(value is! Future);
335 356
336 _Future listeners = _removeListeners(); 357 _FutureListener listeners = _removeListeners();
337 _setValue(value); 358 _setValue(value);
338 _propagateToListeners(this, listeners); 359 _propagateToListeners(this, listeners);
339 } 360 }
340 361
341 void _completeError(error, [StackTrace stackTrace]) { 362 void _completeError(error, [StackTrace stackTrace]) {
342 assert(!_isComplete); 363 assert(!_isComplete);
343 assert(_onValue == null);
344 assert(_onError == null);
345 assert(_whenCompleteAction == null);
346 assert(_errorTest == null);
347 364
348 _Future listeners = _removeListeners(); 365 _FutureListener listeners = _removeListeners();
349 _setError(error, stackTrace); 366 _setError(error, stackTrace);
350 _propagateToListeners(this, listeners); 367 _propagateToListeners(this, listeners);
351 } 368 }
352 369
353 void _asyncComplete(value) { 370 void _asyncComplete(value) {
354 assert(!_isComplete); 371 assert(!_isComplete);
355 assert(_onValue == null);
356 assert(_onError == null);
357 assert(_whenCompleteAction == null);
358 assert(_errorTest == null);
359 // Two corner cases if the value is a future: 372 // Two corner cases if the value is a future:
360 // 1. the future is already completed and an error. 373 // 1. the future is already completed and an error.
361 // 2. the future is not yet completed but might become an error. 374 // 2. the future is not yet completed but might become an error.
362 // The first case means that we must not immediately complete the Future, 375 // The first case means that we must not immediately complete the Future,
363 // as our code would immediately start propagating the error without 376 // as our code would immediately start propagating the error without
364 // giving the time to install error-handlers. 377 // giving the time to install error-handlers.
365 // However the second case requires us to deal with the value immediately. 378 // However the second case requires us to deal with the value immediately.
366 // Otherwise the value could complete with an error and report an 379 // Otherwise the value could complete with an error and report an
367 // unhandled error, even though we know we are already going to listen to 380 // unhandled error, even though we know we are already going to listen to
368 // it. 381 // it.
(...skipping 27 matching lines...) Expand all
396 } 409 }
397 410
398 _markPendingCompletion(); 411 _markPendingCompletion();
399 _zone.scheduleMicrotask(() { 412 _zone.scheduleMicrotask(() {
400 _completeWithValue(value); 413 _completeWithValue(value);
401 }); 414 });
402 } 415 }
403 416
404 void _asyncCompleteError(error, StackTrace stackTrace) { 417 void _asyncCompleteError(error, StackTrace stackTrace) {
405 assert(!_isComplete); 418 assert(!_isComplete);
406 assert(_onValue == null);
407 assert(_onError == null);
408 assert(_whenCompleteAction == null);
409 assert(_errorTest == null);
410 419
411 _markPendingCompletion(); 420 _markPendingCompletion();
412 _zone.scheduleMicrotask(() { 421 _zone.scheduleMicrotask(() {
413 _completeError(error, stackTrace); 422 _completeError(error, stackTrace);
414 }); 423 });
415 } 424 }
416 425
417 /** 426 /**
418 * Propagates the value/error of [source] to its [listeners].
419 *
420 * Unlinks all listeners and propagates the source to each listener
421 * separately.
422 */
423 static void _propagateMultipleListeners(_Future source, _Future listeners) {
424 assert(listeners != null);
425 assert(listeners._nextListener != null);
426 do {
427 _Future listener = listeners;
428 listeners = listener._nextListener;
429 listener._nextListener = null;
430 _propagateToListeners(source, listener);
431 } while (listeners != null);
432 }
433
434 /**
435 * Propagates the value/error of [source] to its [listeners], executing the 427 * Propagates the value/error of [source] to its [listeners], executing the
436 * listeners' callbacks. 428 * listeners' callbacks.
437 * 429 *
438 * If [runCallback] is true (which should be the default) it executes 430 * If [runCallback] is true (which should be the default) it executes
Søren Gjesse 2014/10/09 06:55:51 I don't see the runCallback argument.
Lasse Reichstein Nielsen 2014/10/09 07:53:21 True. I think I removed that earlier, when I start
439 * the registered action of listeners. If it is `false` then the callback is 431 * the registered action of listeners. If it is `false` then the callback is
440 * skipped. This is used to complete futures with chained futures. 432 * skipped. This is used to complete futures with chained futures.
441 */ 433 */
442 static void _propagateToListeners(_Future source, _Future listeners) { 434 static void _propagateToListeners(_Future source, _FutureListener listeners) {
443 while (true) { 435 while (true) {
444 if (!source._isComplete) return; // Chained future. 436 assert(source._isComplete);
445 bool hasError = source._hasError; 437 bool hasError = source._hasError;
446 if (hasError && listeners == null) { 438 if (listeners == null) {
447 AsyncError asyncError = source._error; 439 if (hasError) {
448 source._zone.handleUncaughtError( 440 AsyncError asyncError = source._error;
449 asyncError.error, asyncError.stackTrace); 441 source._zone.handleUncaughtError(
442 asyncError.error, asyncError.stackTrace);
443 }
450 return; 444 return;
451 } 445 }
Søren Gjesse 2014/10/09 06:55:52 I think the comment about special-casing the one l
Lasse Reichstein Nielsen 2014/10/09 07:53:21 Done.
452 if (listeners == null) return; 446 while (listeners._nextListener != null) {
453 _Future listener = listeners; 447 _FutureListener listener = listeners;
454 if (listener._nextListener != null) { 448 listeners = listener._nextListener;
455 // Usually futures only have one listener. If they have several, we 449 listener._nextListener = null;
456 // handle them specially. 450 _propagateToListeners(source, listener);
457 _propagateMultipleListeners(source, listeners);
458 return;
459 } 451 }
452 _FutureListener listener = listeners;
460 // Do the actual propagation. 453 // Do the actual propagation.
461 // Set initial state of listenerHasValue and listenerValueOrError. These 454 // Set initial state of listenerHasValue and listenerValueOrError. These
462 // variables are updated, with the outcome of potential callbacks. 455 // variables are updated, with the outcome of potential callbacks.
463 bool listenerHasValue = true; 456 bool listenerHasValue = true;
464 final sourceValue = source._hasValue ? source._value : null; 457 final sourceValue = hasError ? null : source._value;
465 var listenerValueOrError = sourceValue; 458 var listenerValueOrError = sourceValue;
466 // Set to true if a whenComplete needs to wait for a future. 459 // Set to true if a whenComplete needs to wait for a future.
467 // The whenComplete action will resume the propagation by itself. 460 // The whenComplete action will resume the propagation by itself.
468 bool isPropagationAborted = false; 461 bool isPropagationAborted = false;
469 // TODO(floitsch): mark the listener as pending completion. Currently 462 // TODO(floitsch): mark the listener as pending completion. Currently
470 // we can't do this, since the markPendingCompletion verifies that 463 // we can't do this, since the markPendingCompletion verifies that
471 // the future is not already marked (or chained). 464 // the future is not already marked (or chained).
472 // Only if we either have an error or callbacks, go into this, somewhat 465 // Only if we either have an error or callbacks, go into this, somewhat
473 // expensive, branch. Here we'll enter/leave the zone. Many futures 466 // expensive, branch. Here we'll enter/leave the zone. Many futures
474 // doesn't have callbacks, so this is a significant optimization. 467 // doesn't have callbacks, so this is a significant optimization.
475 if (hasError || 468 if (hasError || (listener.handlesValue || listener.handlesComplete)) {
476 listener._onValue != null ||
477 listener._whenCompleteAction != null) {
478 Zone zone = listener._zone; 469 Zone zone = listener._zone;
479 if (hasError && !source._zone.inSameErrorZone(zone)) { 470 if (hasError && !source._zone.inSameErrorZone(zone)) {
480 // Don't cross zone boundaries with errors. 471 // Don't cross zone boundaries with errors.
481 AsyncError asyncError = source._error; 472 AsyncError asyncError = source._error;
482 source._zone.handleUncaughtError( 473 source._zone.handleUncaughtError(
483 asyncError.error, asyncError.stackTrace); 474 asyncError.error, asyncError.stackTrace);
484 return; 475 return;
485 } 476 }
486 477
487 Zone oldZone; 478 Zone oldZone;
488 if (!identical(Zone.current, zone)) { 479 if (!identical(Zone.current, zone)) {
489 // Change zone if it's not current. 480 // Change zone if it's not current.
490 oldZone = Zone._enter(zone); 481 oldZone = Zone._enter(zone);
491 } 482 }
492 483
493 bool handleValueCallback() { 484 bool handleValueCallback() {
494 try { 485 try {
495 listenerValueOrError = zone.runUnary(listener._onValue, 486 listenerValueOrError = zone.runUnary(listener._onValue,
496 sourceValue); 487 sourceValue);
497 return true; 488 return true;
498 } catch (e, s) { 489 } catch (e, s) {
499 listenerValueOrError = new AsyncError(e, s); 490 listenerValueOrError = new AsyncError(e, s);
500 return false; 491 return false;
501 } 492 }
502 } 493 }
503 494
504 void handleError() { 495 void handleError() {
505 AsyncError asyncError = source._error; 496 AsyncError asyncError = source._error;
506 _FutureErrorTest test = listener._errorTest;
507 bool matchesTest = true; 497 bool matchesTest = true;
508 if (test != null) { 498 if (listener.hasErrorTest) {
499 _FutureErrorTest test = listener._errorTest;
509 try { 500 try {
510 matchesTest = zone.runUnary(test, asyncError.error); 501 matchesTest = zone.runUnary(test, asyncError.error);
511 } catch (e, s) { 502 } catch (e, s) {
512 // TODO(ajohnsen): Should we suport rethrow for test throws?
513 listenerValueOrError = identical(asyncError.error, e) ? 503 listenerValueOrError = identical(asyncError.error, e) ?
514 asyncError : new AsyncError(e, s); 504 asyncError : new AsyncError(e, s);
515 listenerHasValue = false; 505 listenerHasValue = false;
516 return; 506 return;
517 } 507 }
518 } 508 }
519 Function errorCallback = listener._onError; 509 Function errorCallback = listener._onError;
520 if (matchesTest && errorCallback != null) { 510 if (matchesTest && errorCallback != null) {
521 try { 511 try {
522 if (errorCallback is ZoneBinaryCallback) { 512 if (errorCallback is ZoneBinaryCallback) {
(...skipping 22 matching lines...) Expand all
545 var completeResult; 535 var completeResult;
546 try { 536 try {
547 completeResult = zone.run(listener._whenCompleteAction); 537 completeResult = zone.run(listener._whenCompleteAction);
548 } catch (e, s) { 538 } catch (e, s) {
549 if (hasError && identical(source._error.error, e)) { 539 if (hasError && identical(source._error.error, e)) {
550 listenerValueOrError = source._error; 540 listenerValueOrError = source._error;
551 } else { 541 } else {
552 listenerValueOrError = new AsyncError(e, s); 542 listenerValueOrError = new AsyncError(e, s);
553 } 543 }
554 listenerHasValue = false; 544 listenerHasValue = false;
545 return;
555 } 546 }
556 if (completeResult is Future) { 547 if (completeResult is Future) {
557 listener._isChained = true; 548 _Future result = listener.result;
549 result._isChained = true;
558 isPropagationAborted = true; 550 isPropagationAborted = true;
559 completeResult.then((ignored) { 551 completeResult.then((ignored) {
560 // Try again. Since the future is marked as chained it won't run 552 _propagateToListeners(source, new _FutureListener.chain(result));
561 // the whenComplete again.
562 _propagateToListeners(source, listener);
563 }, onError: (error, [stackTrace]) { 553 }, onError: (error, [stackTrace]) {
564 // When there is an error, we have to make the error the new 554 // When there is an error, we have to make the error the new
565 // result of the current listener. 555 // result of the current listener.
566 if (completeResult is! _Future) { 556 if (completeResult is! _Future) {
567 // This should be a rare case. 557 // This should be a rare case.
568 completeResult = new _Future(); 558 completeResult = new _Future();
569 completeResult._setError(error, stackTrace); 559 completeResult._setError(error, stackTrace);
570 } 560 }
571 _propagateToListeners(completeResult, listener); 561 _propagateToListeners(completeResult,
562 new _FutureListener.chain(result));
572 }); 563 });
573 } 564 }
574 } 565 }
575 566
576 if (!hasError) { 567 if (!hasError) {
577 if (listener._onValue != null) { 568 if (listener.handlesValue) {
578 listenerHasValue = handleValueCallback(); 569 listenerHasValue = handleValueCallback();
579 } 570 }
580 } else { 571 } else {
581 handleError(); 572 handleError();
582 } 573 }
583 if (listener._whenCompleteAction != null) { 574 if (listener.handlesComplete) {
584 handleWhenCompleteCallback(); 575 handleWhenCompleteCallback();
585 } 576 }
586 // If we changed zone, oldZone will not be null. 577 // If we changed zone, oldZone will not be null.
587 if (oldZone != null) Zone._leave(oldZone); 578 if (oldZone != null) Zone._leave(oldZone);
588 listener._onValueCallback = null;
589 listener._errorTestCallback = null;
590 listener._onErrorCallback = null;
591 listener._whenCompleteActionCallback = null;
592 579
593 if (isPropagationAborted) return; 580 if (isPropagationAborted) return;
594 // If the listener's value is a future we need to chain it. Note that 581 // If the listener's value is a future we need to chain it. Note that
595 // this can only happen if there is a callback. Since 'is' checks 582 // this can only happen if there is a callback. Since 'is' checks
596 // can be expensive, we're trying to avoid it. 583 // can be expensive, we're trying to avoid it.
597 if (listenerHasValue && 584 if (listenerHasValue &&
598 !identical(sourceValue, listenerValueOrError) && 585 !identical(sourceValue, listenerValueOrError) &&
599 listenerValueOrError is Future) { 586 listenerValueOrError is Future) {
600 Future chainSource = listenerValueOrError; 587 Future chainSource = listenerValueOrError;
601 // Shortcut if the chain-source is already completed. Just continue 588 // Shortcut if the chain-source is already completed. Just continue
602 // the loop. 589 // the loop.
590 _Future result = listener.result;
603 if (chainSource is _Future) { 591 if (chainSource is _Future) {
604 if (chainSource._isComplete) { 592 if (chainSource._isComplete) {
605 // propagate the value (simulating a tail call). 593 // propagate the value (simulating a tail call).
606 listener._isChained = true; 594 result._isChained = true;
607 source = chainSource; 595 source = chainSource;
608 listeners = listener; 596 listeners = new _FutureListener.chain(result);
609 continue; 597 continue;
610 } else { 598 } else {
611 _chainCoreFuture(chainSource, listener); 599 _chainCoreFuture(chainSource, result);
612 } 600 }
613 } else { 601 } else {
614 _chainForeignFuture(chainSource, listener); 602 _chainForeignFuture(chainSource, result);
615 } 603 }
616 return; 604 return;
617 } 605 }
618 } 606 }
607 _Future result = listener.result;
608 listeners = result._removeListeners();
619 if (listenerHasValue) { 609 if (listenerHasValue) {
620 listeners = listener._removeListeners(); 610 result._setValue(listenerValueOrError);
621 listener._setValue(listenerValueOrError);
622 } else { 611 } else {
623 listeners = listener._removeListeners();
624 AsyncError asyncError = listenerValueOrError; 612 AsyncError asyncError = listenerValueOrError;
625 listener._setError(asyncError.error, asyncError.stackTrace); 613 result._setErrorObject(asyncError);
626 } 614 }
627 // Prepare for next round. 615 // Prepare for next round.
628 source = listener; 616 source = result;
629 } 617 }
630 } 618 }
631 619
632 Future timeout(Duration timeLimit, {onTimeout()}) { 620 Future timeout(Duration timeLimit, {onTimeout()}) {
633 if (_isComplete) return new _Future.immediate(this); 621 if (_isComplete) return new _Future.immediate(this);
634 _Future result = new _Future(); 622 _Future result = new _Future();
635 Timer timer; 623 Timer timer;
636 if (onTimeout == null) { 624 if (onTimeout == null) {
637 timer = new Timer(timeLimit, () { 625 timer = new Timer(timeLimit, () {
638 result._completeError(new TimeoutException("Future not completed", 626 result._completeError(new TimeoutException("Future not completed",
(...skipping 17 matching lines...) Expand all
656 } 644 }
657 }, onError: (e, s) { 645 }, onError: (e, s) {
658 if (timer.isActive) { 646 if (timer.isActive) {
659 timer.cancel(); 647 timer.cancel();
660 result._completeError(e, s); 648 result._completeError(e, s);
661 } 649 }
662 }); 650 });
663 return result; 651 return result;
664 } 652 }
665 } 653 }
OLDNEW
« no previous file with comments | « no previous file | sdk/lib/async/schedule_microtask.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698