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

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

Issue 18080015: Revert "Make StreamController be a StreamSink, not just an EventSink." (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Created 7 years, 5 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/broadcast_stream_controller.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 deprecatedFutureValue(_FutureImpl future) => 7 deprecatedFutureValue(_FutureImpl future) =>
8 future._isComplete ? future._resultOrListeners : null; 8 future._isComplete ? future._resultOrListeners : null;
9 9
10 abstract class _Completer<T> implements Completer<T> { 10 abstract class _Completer<T> implements Completer<T> {
(...skipping 31 matching lines...) Expand 10 before | Expand all | Expand 10 after
42 _Zone.current.handleUncaughtError(error); 42 _Zone.current.handleUncaughtError(error);
43 } 43 }
44 } 44 }
45 45
46 bool get isCompleted => _isComplete; 46 bool get isCompleted => _isComplete;
47 } 47 }
48 48
49 class _AsyncCompleter<T> extends _Completer<T> { 49 class _AsyncCompleter<T> extends _Completer<T> {
50 void _setFutureValue(T value) { 50 void _setFutureValue(T value) {
51 _FutureImpl future = this.future; 51 _FutureImpl future = this.future;
52 future._asyncSetValue(value); 52 runAsync(() { future._setValue(value); });
53 } 53 }
54 54
55 void _setFutureError(error) { 55 void _setFutureError(error) {
56 _FutureImpl future = this.future; 56 _FutureImpl future = this.future;
57 future._asyncSetError(error); 57 runAsync(() { future._setError(error); });
58 } 58 }
59 } 59 }
60 60
61 class _SyncCompleter<T> extends _Completer<T> { 61 class _SyncCompleter<T> extends _Completer<T> {
62 void _setFutureValue(T value) { 62 void _setFutureValue(T value) {
63 _FutureImpl future = this.future; 63 _FutureImpl future = this.future;
64 future._setValue(value); 64 future._setValue(value);
65 } 65 }
66 66
67 void _setFutureError(error) { 67 void _setFutureError(error) {
(...skipping 19 matching lines...) Expand all
87 void _sendError(error); 87 void _sendError(error);
88 88
89 bool _inSameErrorZone(_Zone otherZone); 89 bool _inSameErrorZone(_Zone otherZone);
90 } 90 }
91 91
92 /** Adapter for a [_FutureImpl] to be a future result listener. */ 92 /** Adapter for a [_FutureImpl] to be a future result listener. */
93 class _FutureListenerWrapper<T> implements _FutureListener<T> { 93 class _FutureListenerWrapper<T> implements _FutureListener<T> {
94 _FutureImpl future; 94 _FutureImpl future;
95 _FutureListener _nextListener; 95 _FutureListener _nextListener;
96 _FutureListenerWrapper(this.future); 96 _FutureListenerWrapper(this.future);
97 _sendValue(T value) { future._setValueUnchecked(value); } 97 _sendValue(T value) { future._setValue(value); }
98 _sendError(error) { future._setErrorUnchecked(error); } 98 _sendError(error) { future._setError(error); }
99 bool _inSameErrorZone(_Zone otherZone) => future._inSameErrorZone(otherZone); 99 bool _inSameErrorZone(_Zone otherZone) => future._inSameErrorZone(otherZone);
100 } 100 }
101 101
102 /** 102 /**
103 * This listener is installed at error-zone boundaries. It signals an 103 * This listener is installed at error-zone boundaries. It signals an
104 * uncaught error in the zone of origin when an error is sent from one error 104 * uncaught error in the zone of origin when an error is sent from one error
105 * zone to another. 105 * zone to another.
106 * 106 *
107 * When a Future is listening to another Future and they have not been 107 * When a Future is listening to another Future and they have not been
108 * instantiated in the same error-zone then Futures put an instance of this 108 * instantiated in the same error-zone then Futures put an instance of this
(...skipping 46 matching lines...) Expand 10 before | Expand all | Expand 10 after
155 155
156 class _FutureImpl<T> implements Future<T> { 156 class _FutureImpl<T> implements Future<T> {
157 // State of the future. The state determines the interpretation of the 157 // State of the future. The state determines the interpretation of the
158 // [resultOrListeners] field. 158 // [resultOrListeners] field.
159 // TODO(lrn): rename field since it can also contain a chained future. 159 // TODO(lrn): rename field since it can also contain a chained future.
160 160
161 /// Initial state, waiting for a result. In this state, the 161 /// Initial state, waiting for a result. In this state, the
162 /// [resultOrListeners] field holds a single-linked list of 162 /// [resultOrListeners] field holds a single-linked list of
163 /// [FutureListener] listeners. 163 /// [FutureListener] listeners.
164 static const int _INCOMPLETE = 0; 164 static const int _INCOMPLETE = 0;
165 /// Pending completion. Set when completed using [_asyncSetValue] or
166 /// [_asyncSetError]. It is an error to try to complete it again.
167 static const int _PENDING_COMPLETE = 1;
168 /// The future has been chained to another future. The result of that 165 /// The future has been chained to another future. The result of that
169 /// other future becomes the result of this future as well. 166 /// other future becomes the result of this future as well.
170 /// In this state, the [resultOrListeners] field holds the future that 167 /// In this state, the [resultOrListeners] field holds the future that
171 /// will give the result to this future. Both existing and new listeners are 168 /// will give the result to this future. Both existing and new listeners are
172 /// forwarded directly to the other future. 169 /// forwarded directly to the other future.
173 static const int _CHAINED = 2; 170 static const int _CHAINED = 1;
174 /// The future has been chained to another future, but there hasn't been 171 /// The future has been chained to another future, but there hasn't been
175 /// any listeners added to this future yet. If it is completed with an 172 /// any listeners added to this future yet. If it is completed with an
176 /// error, the error will be considered unhandled. 173 /// error, the error will be considered unhandled.
177 static const int _CHAINED_UNLISTENED = 6; 174 static const int _CHAINED_UNLISTENED = 3;
178 /// The future has been completed with a value result. 175 /// The future has been completed with a value result.
179 static const int _VALUE = 8; 176 static const int _VALUE = 4;
180 /// The future has been completed with an error result. 177 /// The future has been completed with an error result.
181 static const int _ERROR = 12; 178 static const int _ERROR = 6;
182 /// Extra bit set when the future has been completed with an error result. 179 /// Extra bit set when the future has been completed with an error result.
183 /// but no listener has been scheduled to receive the error. 180 /// but no listener has been scheduled to receive the error.
184 /// If the bit is still set when a [runAsync] call triggers, the error will 181 /// If the bit is still set when a [runAsync] call triggers, the error will
185 /// be reported to the top-level handler. 182 /// be reported to the top-level handler.
186 /// Assigning a listener before that time will clear the bit. 183 /// Assigning a listener before that time will clear the bit.
187 static const int _UNHANDLED_ERROR = 16; 184 static const int _UNHANDLED_ERROR = 8;
188 185
189 /** Whether the future is complete, and as what. */ 186 /** Whether the future is complete, and as what. */
190 int _state = _INCOMPLETE; 187 int _state = _INCOMPLETE;
191 188
192 final _Zone _zone = _Zone.current.fork(); 189 final _Zone _zone = _Zone.current.fork();
193 190
194 bool get _isChained => (_state & _CHAINED) != 0; 191 bool get _isChained => (_state & _CHAINED) != 0;
195 bool get _hasChainedListener => _state == _CHAINED; 192 bool get _hasChainedListener => _state == _CHAINED;
196 bool get _isComplete => _state >= _VALUE; 193 bool get _isComplete => _state >= _VALUE;
197 bool get _mayComplete => _state == _INCOMPLETE;
198 bool get _hasValue => _state == _VALUE; 194 bool get _hasValue => _state == _VALUE;
199 bool get _hasError => _state >= _ERROR; 195 bool get _hasError => _state >= _ERROR;
200 bool get _hasUnhandledError => _state >= _UNHANDLED_ERROR; 196 bool get _hasUnhandledError => _state >= _UNHANDLED_ERROR;
201 197
202 void _clearUnhandledError() { 198 void _clearUnhandledError() {
203 _state &= ~_UNHANDLED_ERROR; 199 _state &= ~_UNHANDLED_ERROR;
204 } 200 }
205 201
206 /** 202 /**
207 * Either the result, a list of listeners or another future. 203 * Either the result, a list of listeners or another future.
(...skipping 79 matching lines...) Expand 10 before | Expand all | Expand 10 after
287 return new _WhenFuture<T>(action).._subscribeTo(this); 283 return new _WhenFuture<T>(action).._subscribeTo(this);
288 } 284 }
289 285
290 Stream<T> asStream() => new Stream.fromFuture(this); 286 Stream<T> asStream() => new Stream.fromFuture(this);
291 287
292 bool _inSameErrorZone(_Zone otherZone) { 288 bool _inSameErrorZone(_Zone otherZone) {
293 return _zone.inSameErrorZone(otherZone); 289 return _zone.inSameErrorZone(otherZone);
294 } 290 }
295 291
296 void _setValue(T value) { 292 void _setValue(T value) {
297 if (!_mayComplete) throw new StateError("Future already completed"); 293 if (_isComplete) throw new StateError("Future already completed");
298 _setValueUnchecked(value);
299 }
300
301 void _setValueUnchecked(T value) {
302 _FutureListener listeners = _isChained ? null : _removeListeners(); 294 _FutureListener listeners = _isChained ? null : _removeListeners();
303 _state = _VALUE; 295 _state = _VALUE;
304 _resultOrListeners = value; 296 _resultOrListeners = value;
305 while (listeners != null) { 297 while (listeners != null) {
306 _FutureListener listener = listeners; 298 _FutureListener listener = listeners;
307 listeners = listener._nextListener; 299 listeners = listener._nextListener;
308 listener._nextListener = null; 300 listener._nextListener = null;
309 listener._sendValue(value); 301 listener._sendValue(value);
310 } 302 }
311 } 303 }
312 304
313 void _setError(Object error) { 305 void _setError(error) {
314 if (!_mayComplete) throw new StateError("Future already completed"); 306 if (_isComplete) throw new StateError("Future already completed");
315 _setErrorUnchecked(error);
316 }
317 307
318 void _setErrorUnchecked(Object error) {
319 _FutureListener listeners; 308 _FutureListener listeners;
320 bool hasListeners; 309 bool hasListeners;
321 if (_isChained) { 310 if (_isChained) {
322 listeners = null; 311 listeners = null;
323 hasListeners = (_state == _CHAINED); // and not _CHAINED_UNLISTENED. 312 hasListeners = (_state == _CHAINED); // and not _CHAINED_UNLISTENED.
324 } else { 313 } else {
325 listeners = _removeListeners(); 314 listeners = _removeListeners();
326 hasListeners = (listeners != null); 315 hasListeners = (listeners != null);
327 } 316 }
328 317
329 _state = _ERROR; 318 _state = _ERROR;
330 _resultOrListeners = error; 319 _resultOrListeners = error;
331 320
332 if (!hasListeners) { 321 if (!hasListeners) {
333 _scheduleUnhandledError(); 322 _scheduleUnhandledError();
334 return; 323 return;
335 } 324 }
336 while (listeners != null) { 325 while (listeners != null) {
337 _FutureListener listener = listeners; 326 _FutureListener listener = listeners;
338 listeners = listener._nextListener; 327 listeners = listener._nextListener;
339 listener._nextListener = null; 328 listener._nextListener = null;
340 listener._sendError(error); 329 listener._sendError(error);
341 } 330 }
342 } 331 }
343 332
344 void _asyncSetValue(T value) {
345 if (!_mayComplete) throw new StateError("Future already completed");
346 _state = _PENDING_COMPLETE;
347 runAsync(() { _setValueUnchecked(value); });
348 }
349
350 void _asyncSetError(Object error) {
351 if (!_mayComplete) throw new StateError("Future already completed");
352 _state = _PENDING_COMPLETE;
353 runAsync(() { _setErrorUnchecked(error); });
354 }
355
356 void _scheduleUnhandledError() { 333 void _scheduleUnhandledError() {
357 assert(_state == _ERROR); 334 assert(_state == _ERROR);
358 _state = _ERROR | _UNHANDLED_ERROR; 335 _state = _ERROR | _UNHANDLED_ERROR;
359 // Wait for the rest of the current event's duration to see 336 // Wait for the rest of the current event's duration to see
360 // if a subscriber is added to handle the error. 337 // if a subscriber is added to handle the error.
361 runAsync(() { 338 runAsync(() {
362 if (_hasUnhandledError) { 339 if (_hasUnhandledError) {
363 // No error handler has been added since the error was set. 340 // No error handler has been added since the error was set.
364 _clearUnhandledError(); 341 _clearUnhandledError();
365 // TODO(floitsch): Hook this into unhandled error handling. 342 // TODO(floitsch): Hook this into unhandled error handling.
(...skipping 310 matching lines...) Expand 10 before | Expand all | Expand 10 after
676 _setError(error); 653 _setError(error);
677 }, onError: _setError); 654 }, onError: _setError);
678 return; 655 return;
679 } 656 }
680 } catch (e, s) { 657 } catch (e, s) {
681 error = _asyncError(e, s); 658 error = _asyncError(e, s);
682 } 659 }
683 _setError(error); 660 _setError(error);
684 } 661 }
685 } 662 }
OLDNEW
« no previous file with comments | « sdk/lib/async/broadcast_stream_controller.dart ('k') | sdk/lib/async/stream.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698