| OLD | NEW |
| 1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file | 1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file |
| 2 // for details. All rights reserved. Use of this source code is governed by a | 2 // for details. All rights reserved. Use of this source code is governed by a |
| 3 // BSD-style license that can be found in the LICENSE file. | 3 // BSD-style license that can be found in the LICENSE file. |
| 4 | 4 |
| 5 part of dart.async; | 5 part of dart.async; |
| 6 | 6 |
| 7 /** 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 Loading... |
| 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 Loading... |
| 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 Loading... |
| 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 Loading... |
| 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 Loading... |
| 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 Loading... |
| 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 } |
| OLD | NEW |