Chromium Code Reviews| Index: sdk/lib/async/stream_impl.dart |
| diff --git a/sdk/lib/async/stream_impl.dart b/sdk/lib/async/stream_impl.dart |
| index be661e7bd5eded48d8df6c4fe43c9b88ee44fdf4..d721ae82b48cd4fd8eedc44efca39d79148de8e7 100644 |
| --- a/sdk/lib/async/stream_impl.dart |
| +++ b/sdk/lib/async/stream_impl.dart |
| @@ -25,7 +25,7 @@ const int _STREAM_FIRING = 8; |
| const int _STREAM_CALLBACK = 16; |
| /// The count of times a stream has paused is stored in the |
| /// state, shifted by this amount. |
| -const int _STREAM_PAUSE_COUNT_SHIFT = 8; |
| +const int _STREAM_PAUSE_COUNT_SHIFT = 5; |
| // States for listeners. |
| @@ -148,6 +148,9 @@ abstract class _StreamImpl<T> extends Stream<T> { |
| /** Whether one or more active subscribers have requested a pause. */ |
| bool get _isPaused => _state >= (1 << _STREAM_PAUSE_COUNT_SHIFT); |
| + /** Whether we are currently executing a state-chance callback. */ |
| + bool get _isInCallback => (_state & _STREAM_CALLBACK) != 0; |
| + |
| /** Check whether the pending event queue is non-empty */ |
| bool get _hasPendingEvent => |
| _pendingEvents != null && !_pendingEvents.isEmpty; |
| @@ -231,7 +234,6 @@ abstract class _StreamImpl<T> extends Stream<T> { |
| void _endFiring() { |
| assert(_isFiring); |
| _state ^= _STREAM_FIRING; |
| - |
| if (!_hasSubscribers) { |
| _callOnSubscriptionStateChange(); |
| } else if (_isPaused) { |
| @@ -332,16 +334,24 @@ abstract class _StreamImpl<T> extends Stream<T> { |
| void _callOnPauseStateChange() { |
| // After calling [_close], all pauses are handled internally by the Stream. |
| if (_isClosed) return; |
| - _state |= _STREAM_CALLBACK; |
| + if (!_isInCallback) { |
| + _state |= _STREAM_CALLBACK; |
| + _onPauseStateChange(); |
| + _state ^= _STREAM_CALLBACK; |
| + return; |
| + } |
|
Anders Johnsen
2013/02/20 13:17:32
Use else.
|
| _onPauseStateChange(); |
| - _state ^= _STREAM_CALLBACK; |
| } |
| /** Calls [_onSubscriptionStateChange] while setting callback bit. */ |
| void _callOnSubscriptionStateChange() { |
| - _state |= _STREAM_CALLBACK; |
| + if (!_isInCallback) { |
| + _state |= _STREAM_CALLBACK; |
| + _onSubscriptionStateChange(); |
| + _state ^= _STREAM_CALLBACK; |
| + return; |
| + } |
| _onSubscriptionStateChange(); |
| - _state ^= _STREAM_CALLBACK; |
| } |
| /** |
| @@ -471,9 +481,12 @@ class _SingleStreamImpl<T> extends _StreamImpl<T> { |
| bool get _isPaused => (!_hasSubscribers && !_isComplete) || super._isPaused; |
| - // A single-stream is considered paused when it has no subscriber. |
| - bool get _canFireEvent => |
| - _mayFireState && !_hasPendingEvent && _hasSubscribers; |
| + bool get _canFireEvent { |
| + // A single-stream is considered paused when it has no subscriber and |
| + // isn't complete, so it can't fire if there is no subscriber. |
| + // It won't try to fire when completed, so no need to check that. |
| + return _mayFireState && !_hasPendingEvent && _hasSubscribers; |
| + } |
| /** Whether there is currently a subscriber on this [Stream]. */ |