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

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

Issue 17064008: Revert "Zone support for Futures." (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Created 7 years, 6 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/stream_controller.dart ('k') | sdk/lib/async/timer.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 /** Throws the given error in the next cycle. */
8 _throwDelayed(var error, [Object stackTrace]) {
9 // We are going to reach the top-level here, but there might be a global
10 // exception handler. This means that we shouldn't print the stack trace.
11 // TODO(floitsch): Find better solution that doesn't print the stack trace
12 // if there is a global exception handler.
13 runAsync(() {
14 if (stackTrace != null) print(stackTrace);
15 var trace = getAttachedStackTrace(error);
16 if (trace != null && trace != stackTrace) print(trace);
17 throw error;
18 });
19 }
20
7 /** Abstract and private interface for a place to put events. */ 21 /** Abstract and private interface for a place to put events. */
8 abstract class _EventSink<T> { 22 abstract class _EventSink<T> {
9 void _add(T data); 23 void _add(T data);
10 void _addError(Object error); 24 void _addError(Object error);
11 void _close(); 25 void _close();
12 } 26 }
13 27
14 /** 28 /**
15 * Abstract and private interface for a place to send events. 29 * Abstract and private interface for a place to send events.
16 * 30 *
(...skipping 58 matching lines...) Expand 10 before | Expand all | Expand 10 after
75 static const int _STATE_HAS_PENDING = 32; 89 static const int _STATE_HAS_PENDING = 32;
76 static const int _STATE_PAUSE_COUNT = 64; 90 static const int _STATE_PAUSE_COUNT = 64;
77 static const int _STATE_PAUSE_COUNT_SHIFT = 6; 91 static const int _STATE_PAUSE_COUNT_SHIFT = 6;
78 92
79 /* Event handlers provided in constructor. */ 93 /* Event handlers provided in constructor. */
80 /* TODO(7733): Fix Function->_DataHandler<T> when dart2js understands 94 /* TODO(7733): Fix Function->_DataHandler<T> when dart2js understands
81 * parameterized function types. */ 95 * parameterized function types. */
82 Function _onData; 96 Function _onData;
83 _ErrorHandler _onError; 97 _ErrorHandler _onError;
84 _DoneHandler _onDone; 98 _DoneHandler _onDone;
85 final _Zone _zone = _Zone.current;
86 99
87 /** Bit vector based on state-constants above. */ 100 /** Bit vector based on state-constants above. */
88 int _state; 101 int _state;
89 102
90 /** 103 /**
91 * Queue of pending events. 104 * Queue of pending events.
92 * 105 *
93 * Is created when necessary, or set in constructor for preconfigured events. 106 * Is created when necessary, or set in constructor for preconfigured events.
94 */ 107 */
95 _PendingEvents _pending; 108 _PendingEvents _pending;
96 109
97 _BufferingStreamSubscription(this._onData, 110 _BufferingStreamSubscription(this._onData,
98 this._onError, 111 this._onError,
99 this._onDone, 112 this._onDone,
100 bool cancelOnError) 113 bool cancelOnError)
101 : _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) { 114 : _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) {
102 assert(_onData != null); 115 assert(_onData != null);
103 assert(_onError != null); 116 assert(_onError != null);
104 assert(_onDone != null); 117 assert(_onDone != null);
105 _zone.expectCallback();
106 } 118 }
107 119
108 /** 120 /**
109 * Sets the subscription's pending events object. 121 * Sets the subscription's pending events object.
110 * 122 *
111 * This can only be done once. The pending events object is used for the 123 * This can only be done once. The pending events object is used for the
112 * rest of the subscription's life cycle. 124 * rest of the subscription's life cycle.
113 */ 125 */
114 void _setPendingEvents(_PendingEvents pendingEvents) { 126 void _setPendingEvents(_PendingEvents pendingEvents) {
115 assert(_pending == null); 127 assert(_pending == null);
(...skipping 98 matching lines...) Expand 10 before | Expand all | Expand 10 after
214 bool get _isPaused => _state >= _STATE_PAUSE_COUNT; 226 bool get _isPaused => _state >= _STATE_PAUSE_COUNT;
215 bool get _canFire => _state < _STATE_IN_CALLBACK; 227 bool get _canFire => _state < _STATE_IN_CALLBACK;
216 bool get _mayResumeInput => 228 bool get _mayResumeInput =>
217 !_isPaused && (_pending == null || _pending.isEmpty); 229 !_isPaused && (_pending == null || _pending.isEmpty);
218 bool get _cancelOnError => (_state & _STATE_CANCEL_ON_ERROR) != 0; 230 bool get _cancelOnError => (_state & _STATE_CANCEL_ON_ERROR) != 0;
219 231
220 bool get isPaused => _isPaused; 232 bool get isPaused => _isPaused;
221 233
222 void _cancel() { 234 void _cancel() {
223 _state |= _STATE_CANCELED; 235 _state |= _STATE_CANCELED;
224 _zone.cancelCallbackExpectation();
225 if (_hasPending) { 236 if (_hasPending) {
226 _pending.cancelSchedule(); 237 _pending.cancelSchedule();
227 } 238 }
228 } 239 }
229 240
230 /** 241 /**
231 * Increment the pause count. 242 * Increment the pause count.
232 * 243 *
233 * Also marks input as paused. 244 * Also marks input as paused.
234 */ 245 */
(...skipping 40 matching lines...) Expand 10 before | Expand all | Expand 10 after
275 _state |= _STATE_CLOSED; 286 _state |= _STATE_CLOSED;
276 if (_canFire) { 287 if (_canFire) {
277 _sendDone(); 288 _sendDone();
278 } else { 289 } else {
279 _addPending(const _DelayedDone()); 290 _addPending(const _DelayedDone());
280 } 291 }
281 } 292 }
282 293
283 // Hooks called when the input is paused, unpaused or canceled. 294 // Hooks called when the input is paused, unpaused or canceled.
284 // These must not throw. If overwritten to call user code, include suitable 295 // These must not throw. If overwritten to call user code, include suitable
285 // try/catch wrapping and send any errors to 296 // try/catch wrapping and send any errors to [_throwDelayed].
286 // [_Zone.current.handleUncaughtError].
287 void _onPause() { 297 void _onPause() {
288 assert(_isInputPaused); 298 assert(_isInputPaused);
289 } 299 }
290 300
291 void _onResume() { 301 void _onResume() {
292 assert(!_isInputPaused); 302 assert(!_isInputPaused);
293 } 303 }
294 304
295 void _onCancel() { 305 void _onCancel() {
296 assert(_isCanceled); 306 assert(_isCanceled);
(...skipping 20 matching lines...) Expand all
317 } 327 }
318 328
319 /* _EventDispatch interface. */ 329 /* _EventDispatch interface. */
320 330
321 void _sendData(T data) { 331 void _sendData(T data) {
322 assert(!_isCanceled); 332 assert(!_isCanceled);
323 assert(!_isPaused); 333 assert(!_isPaused);
324 assert(!_inCallback); 334 assert(!_inCallback);
325 bool wasInputPaused = _isInputPaused; 335 bool wasInputPaused = _isInputPaused;
326 _state |= _STATE_IN_CALLBACK; 336 _state |= _STATE_IN_CALLBACK;
327 _zone.executePeriodicCallbackGuarded(() => _onData(data)); 337 try {
338 _onData(data);
339 } catch (e, s) {
340 _throwDelayed(e, s);
341 }
328 _state &= ~_STATE_IN_CALLBACK; 342 _state &= ~_STATE_IN_CALLBACK;
329 _checkState(wasInputPaused); 343 _checkState(wasInputPaused);
330 } 344 }
331 345
332 void _sendError(var error) { 346 void _sendError(var error) {
333 assert(!_isCanceled); 347 assert(!_isCanceled);
334 assert(!_isPaused); 348 assert(!_isPaused);
335 assert(!_inCallback); 349 assert(!_inCallback);
336 bool wasInputPaused = _isInputPaused; 350 bool wasInputPaused = _isInputPaused;
337 _state |= _STATE_IN_CALLBACK; 351 _state |= _STATE_IN_CALLBACK;
338 if (!_zone.inSameErrorZone(_Zone.current)) { 352 try {
339 // Errors are not allowed to traverse zone boundaries. 353 _onError(error);
340 _Zone.current.handleUncaughtError(error); 354 } catch (e, s) {
341 } else { 355 _throwDelayed(e, s);
342 _zone.executePeriodicCallbackGuarded(() => _onError(error));
343 } 356 }
344 _state &= ~_STATE_IN_CALLBACK; 357 _state &= ~_STATE_IN_CALLBACK;
345 if (_cancelOnError) { 358 if (_cancelOnError) {
346 _cancel(); 359 _cancel();
347 } 360 }
348 _checkState(wasInputPaused); 361 _checkState(wasInputPaused);
349 } 362 }
350 363
351 void _sendDone() { 364 void _sendDone() {
352 assert(!_isCanceled); 365 assert(!_isCanceled);
353 assert(!_isPaused); 366 assert(!_isPaused);
354 assert(!_inCallback); 367 assert(!_inCallback);
355 _state |= (_STATE_CANCELED | _STATE_CLOSED | _STATE_IN_CALLBACK); 368 _state |= (_STATE_CANCELED | _STATE_CLOSED | _STATE_IN_CALLBACK);
356 _zone.executeCallbackGuarded(_onDone); 369 try {
370 _onDone();
371 } catch (e, s) {
372 _throwDelayed(e, s);
373 }
357 _onCancel(); // No checkState after cancel, it is always the last event. 374 _onCancel(); // No checkState after cancel, it is always the last event.
358 _state &= ~_STATE_IN_CALLBACK; 375 _state &= ~_STATE_IN_CALLBACK;
359 } 376 }
360 377
361 /** 378 /**
362 * Call a hook function. 379 * Call a hook function.
363 * 380 *
364 * The call is properly wrapped in code to avoid other callbacks 381 * The call is properly wrapped in code to avoid other callbacks
365 * during the call, and it checks for state changes after the call 382 * during the call, and it checks for state changes after the call
366 * that should cause further callbacks. 383 * that should cause further callbacks.
(...skipping 160 matching lines...) Expand 10 before | Expand all | Expand 10 after
527 typedef void _DataHandler<T>(T value); 544 typedef void _DataHandler<T>(T value);
528 typedef void _ErrorHandler(error); 545 typedef void _ErrorHandler(error);
529 typedef void _DoneHandler(); 546 typedef void _DoneHandler();
530 547
531 548
532 /** Default data handler, does nothing. */ 549 /** Default data handler, does nothing. */
533 void _nullDataHandler(var value) {} 550 void _nullDataHandler(var value) {}
534 551
535 /** Default error handler, reports the error to the global handler. */ 552 /** Default error handler, reports the error to the global handler. */
536 void _nullErrorHandler(error) { 553 void _nullErrorHandler(error) {
537 _Zone.current.handleUncaughtError(error); 554 _throwDelayed(error);
538 } 555 }
539 556
540 /** Default done handler, does nothing. */ 557 /** Default done handler, does nothing. */
541 void _nullDoneHandler() {} 558 void _nullDoneHandler() {}
542 559
543 560
544 /** A delayed event on a buffering stream subscription. */ 561 /** A delayed event on a buffering stream subscription. */
545 abstract class _DelayedEvent { 562 abstract class _DelayedEvent {
546 /** Added as a linked list on the [StreamController]. */ 563 /** Added as a linked list on the [StreamController]. */
547 _DelayedEvent next; 564 _DelayedEvent next;
(...skipping 327 matching lines...) Expand 10 before | Expand all | Expand 10 after
875 _FutureImpl<bool> hasNext = _futureOrPrefetch; 892 _FutureImpl<bool> hasNext = _futureOrPrefetch;
876 _clear(); 893 _clear();
877 hasNext._setValue(false); 894 hasNext._setValue(false);
878 return; 895 return;
879 } 896 }
880 _subscription.pause(); 897 _subscription.pause();
881 _futureOrPrefetch = null; 898 _futureOrPrefetch = null;
882 _state = _STATE_EXTRA_DONE; 899 _state = _STATE_EXTRA_DONE;
883 } 900 }
884 } 901 }
OLDNEW
« no previous file with comments | « sdk/lib/async/stream_controller.dart ('k') | sdk/lib/async/timer.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698