| 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 /** Abstract and private interface for a place to put events. */ | 7 /** Abstract and private interface for a place to put events. */ |
| 8 abstract class _EventSink<T> { | 8 abstract class _EventSink<T> { |
| 9 void _add(T data); | 9 void _add(T data); |
| 10 void _addError(Object error, StackTrace stackTrace); | 10 void _addError(Object error, StackTrace stackTrace); |
| (...skipping 325 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 336 assert(!_isCanceled); | 336 assert(!_isCanceled); |
| 337 assert(!_isPaused); | 337 assert(!_isPaused); |
| 338 assert(!_inCallback); | 338 assert(!_inCallback); |
| 339 bool wasInputPaused = _isInputPaused; | 339 bool wasInputPaused = _isInputPaused; |
| 340 _state |= _STATE_IN_CALLBACK; | 340 _state |= _STATE_IN_CALLBACK; |
| 341 _zone.runUnaryGuarded(_onData, data); | 341 _zone.runUnaryGuarded(_onData, data); |
| 342 _state &= ~_STATE_IN_CALLBACK; | 342 _state &= ~_STATE_IN_CALLBACK; |
| 343 _checkState(wasInputPaused); | 343 _checkState(wasInputPaused); |
| 344 } | 344 } |
| 345 | 345 |
| 346 void _sendError(var error, StackTrace stackTrace) { | 346 void _sendError(Object error, StackTrace stackTrace) { |
| 347 assert(!_isCanceled); | 347 assert(!_isCanceled); |
| 348 assert(!_isPaused); | 348 assert(!_isPaused); |
| 349 assert(!_inCallback); | 349 assert(!_inCallback); |
| 350 bool wasInputPaused = _isInputPaused; | 350 bool wasInputPaused = _isInputPaused; |
| 351 | 351 |
| 352 void sendError() { | 352 void sendError() { |
| 353 // If the subscription has been canceled while waiting for the cancel | 353 // If the subscription has been canceled while waiting for the cancel |
| 354 // future to finish we must not report the error. | 354 // future to finish we must not report the error. |
| 355 if (_isCanceled && !_waitsForCancel) return; | 355 if (_isCanceled && !_waitsForCancel) return; |
| 356 _state |= _STATE_IN_CALLBACK; | 356 _state |= _STATE_IN_CALLBACK; |
| (...skipping 114 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 471 bool cancelOnError }) { | 471 bool cancelOnError }) { |
| 472 cancelOnError = identical(true, cancelOnError); | 472 cancelOnError = identical(true, cancelOnError); |
| 473 StreamSubscription subscription = | 473 StreamSubscription subscription = |
| 474 _createSubscription(onData, onError, onDone, cancelOnError); | 474 _createSubscription(onData, onError, onDone, cancelOnError); |
| 475 _onListen(subscription); | 475 _onListen(subscription); |
| 476 return subscription; | 476 return subscription; |
| 477 } | 477 } |
| 478 | 478 |
| 479 // ------------------------------------------------------------------- | 479 // ------------------------------------------------------------------- |
| 480 /** Create a subscription object. Called by [subcribe]. */ | 480 /** Create a subscription object. Called by [subcribe]. */ |
| 481 _BufferingStreamSubscription<T> _createSubscription( | 481 StreamSubscription<T> _createSubscription( |
| 482 void onData(T data), | 482 void onData(T data), |
| 483 Function onError, | 483 Function onError, |
| 484 void onDone(), | 484 void onDone(), |
| 485 bool cancelOnError) { | 485 bool cancelOnError) { |
| 486 return new _BufferingStreamSubscription<T>(onData, onError, onDone, | 486 return new _BufferingStreamSubscription<T>(onData, onError, onDone, |
| 487 cancelOnError); | 487 cancelOnError); |
| 488 } | 488 } |
| 489 | 489 |
| 490 /** Hook called when the subscription has been created. */ | 490 /** Hook called when the subscription has been created. */ |
| 491 void _onListen(StreamSubscription subscription) {} | 491 void _onListen(StreamSubscription subscription) {} |
| 492 } | 492 } |
| 493 | 493 |
| 494 typedef _PendingEvents _EventGenerator(); | 494 typedef _PendingEvents _EventGenerator(); |
| 495 | 495 |
| 496 /** Stream that generates its own events. */ | 496 /** Stream that generates its own events. */ |
| 497 class _GeneratedStreamImpl<T> extends _StreamImpl<T> { | 497 class _GeneratedStreamImpl<T> extends _StreamImpl<T> { |
| 498 final _EventGenerator _pending; | 498 final _EventGenerator _pending; |
| 499 bool _isUsed = false; | 499 bool _isUsed = false; |
| 500 /** | 500 /** |
| 501 * Initializes the stream to have only the events provided by a | 501 * Initializes the stream to have only the events provided by a |
| 502 * [_PendingEvents]. | 502 * [_PendingEvents]. |
| 503 * | 503 * |
| 504 * A new [_PendingEvents] must be generated for each listen. | 504 * A new [_PendingEvents] must be generated for each listen. |
| 505 */ | 505 */ |
| 506 _GeneratedStreamImpl(this._pending); | 506 _GeneratedStreamImpl(this._pending); |
| 507 | 507 |
| 508 StreamSubscription _createSubscription( | 508 StreamSubscription<T> _createSubscription( |
| 509 void onData(T data), | 509 void onData(T data), |
| 510 Function onError, | 510 Function onError, |
| 511 void onDone(), | 511 void onDone(), |
| 512 bool cancelOnError) { | 512 bool cancelOnError) { |
| 513 if (_isUsed) throw new StateError("Stream has already been listened to."); | 513 if (_isUsed) throw new StateError("Stream has already been listened to."); |
| 514 _isUsed = true; | 514 _isUsed = true; |
| 515 return new _BufferingStreamSubscription( | 515 return new _BufferingStreamSubscription( |
| 516 onData, onError, onDone, cancelOnError).._setPendingEvents(_pending()); | 516 onData, onError, onDone, cancelOnError).._setPendingEvents(_pending()); |
| 517 } | 517 } |
| 518 } | 518 } |
| (...skipping 54 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 573 /** Default error handler, reports the error to the current zone's handler. */ | 573 /** Default error handler, reports the error to the current zone's handler. */ |
| 574 void _nullErrorHandler(error, [StackTrace stackTrace]) { | 574 void _nullErrorHandler(error, [StackTrace stackTrace]) { |
| 575 Zone.current.handleUncaughtError(error, stackTrace); | 575 Zone.current.handleUncaughtError(error, stackTrace); |
| 576 } | 576 } |
| 577 | 577 |
| 578 /** Default done handler, does nothing. */ | 578 /** Default done handler, does nothing. */ |
| 579 void _nullDoneHandler() {} | 579 void _nullDoneHandler() {} |
| 580 | 580 |
| 581 | 581 |
| 582 /** A delayed event on a buffering stream subscription. */ | 582 /** A delayed event on a buffering stream subscription. */ |
| 583 abstract class _DelayedEvent { | 583 abstract class _DelayedEvent<T> { |
| 584 /** Added as a linked list on the [StreamController]. */ | 584 /** Added as a linked list on the [StreamController]. */ |
| 585 _DelayedEvent next; | 585 _DelayedEvent next; |
| 586 /** Execute the delayed event on the [StreamController]. */ | 586 /** Execute the delayed event on the [StreamController]. */ |
| 587 void perform(_EventDispatch dispatch); | 587 void perform(_EventDispatch<T> dispatch); |
| 588 } | 588 } |
| 589 | 589 |
| 590 /** A delayed data event. */ | 590 /** A delayed data event. */ |
| 591 class _DelayedData<T> extends _DelayedEvent { | 591 class _DelayedData<T> extends _DelayedEvent<T> { |
| 592 final T value; | 592 final T value; |
| 593 _DelayedData(this.value); | 593 _DelayedData(this.value); |
| 594 void perform(_EventDispatch<T> dispatch) { | 594 void perform(_EventDispatch<T> dispatch) { |
| 595 dispatch._sendData(value); | 595 dispatch._sendData(value); |
| 596 } | 596 } |
| 597 } | 597 } |
| 598 | 598 |
| 599 /** A delayed error event. */ | 599 /** A delayed error event. */ |
| 600 class _DelayedError extends _DelayedEvent { | 600 class _DelayedError extends _DelayedEvent { |
| 601 final error; | 601 final error; |
| (...skipping 285 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 887 class _BroadcastSubscriptionWrapper<T> implements StreamSubscription<T> { | 887 class _BroadcastSubscriptionWrapper<T> implements StreamSubscription<T> { |
| 888 final _AsBroadcastStream _stream; | 888 final _AsBroadcastStream _stream; |
| 889 | 889 |
| 890 _BroadcastSubscriptionWrapper(this._stream); | 890 _BroadcastSubscriptionWrapper(this._stream); |
| 891 | 891 |
| 892 void onData(void handleData(T data)) { | 892 void onData(void handleData(T data)) { |
| 893 throw new UnsupportedError( | 893 throw new UnsupportedError( |
| 894 "Cannot change handlers of asBroadcastStream source subscription."); | 894 "Cannot change handlers of asBroadcastStream source subscription."); |
| 895 } | 895 } |
| 896 | 896 |
| 897 void onError(void handleError(Object data)) { | 897 void onError(Function handleError) { |
| 898 throw new UnsupportedError( | 898 throw new UnsupportedError( |
| 899 "Cannot change handlers of asBroadcastStream source subscription."); | 899 "Cannot change handlers of asBroadcastStream source subscription."); |
| 900 } | 900 } |
| 901 | 901 |
| 902 void onDone(void handleDone()) { | 902 void onDone(void handleDone()) { |
| 903 throw new UnsupportedError( | 903 throw new UnsupportedError( |
| 904 "Cannot change handlers of asBroadcastStream source subscription."); | 904 "Cannot change handlers of asBroadcastStream source subscription."); |
| 905 } | 905 } |
| 906 | 906 |
| 907 void pause([Future resumeSignal]) { | 907 void pause([Future resumeSignal]) { |
| (...skipping 165 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 1073 _Future<bool> hasNext = _futureOrPrefetch; | 1073 _Future<bool> hasNext = _futureOrPrefetch; |
| 1074 _clear(); | 1074 _clear(); |
| 1075 hasNext._complete(false); | 1075 hasNext._complete(false); |
| 1076 return; | 1076 return; |
| 1077 } | 1077 } |
| 1078 _subscription.pause(); | 1078 _subscription.pause(); |
| 1079 _futureOrPrefetch = null; | 1079 _futureOrPrefetch = null; |
| 1080 _state = _STATE_EXTRA_DONE; | 1080 _state = _STATE_EXTRA_DONE; |
| 1081 } | 1081 } |
| 1082 } | 1082 } |
| OLD | NEW |