Do not allow streams to send events during state-change callbacks.

sdk/lib/async/stream_impl.dart

 // Copyright (c) 2012, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 part of dart.async;
 
 // States shared by single/multi stream implementations.
 
 /// Initial and default state where the stream can receive and send events.
 const int _STREAM_OPEN = 0;
 /// The stream has received a request to complete, but hasn't done so yet.
 /// No further events can be added to the stream.
 const int _STREAM_CLOSED = 1;
 /// The stream has completed and will no longer receive or send events.
 /// Also counts as closed. The stream must not be paused when it's completed.
 /// Always used in conjunction with [_STREAM_CLOSED].
 const int _STREAM_COMPLETE = 2;
 /// Bit that alternates between events, and listeners are updated to the
 /// current value when they are notified of the event.
 const int _STREAM_EVENT_ID = 4;
 const int _STREAM_EVENT_ID_SHIFT = 2;
 /// Bit set while firing and clear while not.
 const int _STREAM_FIRING = 8;
+/// Bit set while calling a pause-state or subscription-state change callback.
+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 = 4;
+const int _STREAM_PAUSE_COUNT_SHIFT = 8;
 
 // States for listeners.
 
 /// The listener is currently not subscribed to its source stream.
 const int _LISTENER_UNSUBSCRIBED = 0;
 /// The listener is actively subscribed to its source stream.
 const int _LISTENER_SUBSCRIBED = 1;
 /// The listener is subscribed until it has been notified of the current event.
 /// This flag bit is always used in conjuction with [_LISTENER_SUBSCRIBED].
 const int _LISTENER_PENDING_UNSUBSCRIBE = 2;
@@ skipping 109 matching lines @@
   /** Whether one or more active subscribers have requested a pause. */
   bool get _isPaused => _state >= (1 << _STREAM_PAUSE_COUNT_SHIFT);
 
   /** Check whether the pending event queue is non-empty */
   bool get _hasPendingEvent =>
       _pendingEvents != null && !_pendingEvents.isEmpty;
 
   /** Whether we are currently firing an event. */
   bool get _isFiring => (_state & _STREAM_FIRING) != 0;
 
+  /** Whether the state bits allow firing. */
+  bool get _mayFireState {
+    // The state disallows firing if:
+    // - an event is currently firing
+    // - a stat-change callback is being called
+    // - the pause-count is not zero.
+    const int mask =
+        _STREAM_FIRING |
+        _STREAM_CALLBACK |
+        ~((1 << _STREAM_PAUSE_COUNT_SHIFT) - 1);
+    return (_state & mask) == 0;
+  }
+
   int get _currentEventIdBit =>
       (_state & _STREAM_EVENT_ID ) >> _STREAM_EVENT_ID_SHIFT;
 
   /** Whether there is currently a subscriber on this [Stream]. */
   bool get _hasSubscribers;
 
   /** Whether the stream can fire a new event. */
-  bool get _canFireEvent => !_isFiring && !_isPaused && !_hasPendingEvent;
+  bool get _canFireEvent => _mayFireState && !_hasPendingEvent;
 
   // State modification.
 
   /** Record an increases in the number of times the listener has paused. */
   void _incrementPauseCount(_StreamListener<T> listener) {
     listener._incrementPauseCount();
     _updatePauseCount(1);
   }
 
   /** Record a decrease in the number of times the listener has paused. */
@@ skipping 35 matching lines @@
     // This sets the _STREAM_FIRING bit and toggles the _STREAM_EVENT_ID
     // bit. All current subscribers will now have a _LISTENER_EVENT_ID
     // that doesn't match _STREAM_EVENT_ID, and they will receive the
     // event being fired.
     _state ^= _STREAM_FIRING | _STREAM_EVENT_ID;
   }
 
   void _endFiring() {
     assert(_isFiring);
     _state ^= _STREAM_FIRING;
-    if (_isPaused) _onPauseStateChange();
-    if (!_hasSubscribers) _onSubscriptionStateChange();
+

[Anders Johnsen, 2013/02/20 12:42:02]

Extra space.

+    if (!_hasSubscribers) {
+      _callOnSubscriptionStateChange();
+    } else if (_isPaused) {
+      _callOnPauseStateChange();
+    }
   }
 
   /**
    * Record that a listener wants a pause from events.
    *
    * This methods is called from [_StreamListener.pause()].
    * Subclasses can override this method, along with [isPaused] and
    * [createSubscription], if they want to do a different handling of paused
    * subscriptions, e.g., a filtering stream pausing its own source if all its
    * subscribers are paused.
    */
   void _pause(_StreamListener<T> listener, Future resumeSignal) {
     assert(identical(listener._source, this));
     if (!listener._isSubscribed) {
       throw new StateError("Subscription has been canceled.");
     }
     assert(!_isComplete);  // There can be no subscribers when complete.
     bool wasPaused = _isPaused;
     _incrementPauseCount(listener);
     if (resumeSignal != null) {
       resumeSignal.whenComplete(() { this._resume(listener, true); });
     }
     if (!wasPaused && !_isFiring) {
-      _onPauseStateChange();
+      _callOnPauseStateChange();
     }
   }
 
   /** Stops pausing due to one request from the given listener. */
   void _resume(_StreamListener<T> listener, bool fromEvent) {
     if (!listener.isPaused) return;
     assert(listener._isSubscribed);
     assert(_isPaused);
     _decrementPauseCount(listener);
     if (!_isPaused) {
-      if (!_isFiring) _onPauseStateChange();
+      if (!_isFiring) _callOnPauseStateChange();
       if (_hasPendingEvent) {
         // If we can fire events now, fire any pending events right away.
         if (fromEvent && !_isFiring) {
           _handlePendingEvents();
         } else {
           _schedulePendingEvents();
         }
       }
     }
   }
@@ skipping 36 matching lines @@
    * Any change in the pause state is only reported after all subscribers have
    * received the event.
    *
    * The [action] must not throw, or the controller will be left in an
    * invalid state.
    *
    * This method must not be called while [isFiring] is true.
    */
   void _forEachSubscriber(void action(_StreamSubscriptionImpl<T> subscription));
 
+  /** Calls [_onPauseStateChange] while setting callback bit. */
+  void _callOnPauseStateChange() {
+    // After calling [_close], all pauses are handled internally by the Stream.
+    if (_isClosed) return;
+    _state |= _STREAM_CALLBACK;
+    _onPauseStateChange();
+    _state ^= _STREAM_CALLBACK;
+  }
+
+  /** Calls [_onSubscriptionStateChange] while setting callback bit. */
+  void _callOnSubscriptionStateChange() {
+    _state |= _STREAM_CALLBACK;
+    _onSubscriptionStateChange();
+    _state ^= _STREAM_CALLBACK;
+  }
+
   /**
    * Called when the first subscriber requests a pause or the last a resume.
    *
    * Read [isPaused] to see the new state.
    */
   void _onPauseStateChange() {}
 
   /**
    * Called when the first listener subscribes or the last unsubscribes.
    *
@@ skipping 101 matching lines @@
  * * [_onPauseStateChange]: Called when entering or leaving paused mode.
  * * [_hasSubscribers]: Test whether there are currently any subscribers.
  * * [_isPaused]: Test whether the stream is currently paused.
  * The user should not add new events while the stream is paused, but if it
  * happens anyway, the stream will enqueue the events just as when new events
  * arrive while still firing an old event.
  */
 class _SingleStreamImpl<T> extends _StreamImpl<T> {
   _StreamListener _subscriber = null;
 
-  /** Whether one or more active subscribers have requested a pause. */
+  // A single-stream is considered paused when it has no subscriber.
+  // Exception is when it's complete (which only matters for pause-state-change
+  // callbacks), where it's not considered paused.
   bool get _isPaused => (!_hasSubscribers && !_isComplete) || super._isPaused;
 
+
+  // A single-stream is considered paused when it has no subscriber.

[Anders Johnsen, 2013/02/20 12:42:02]

Comment.

+  bool get _canFireEvent =>
+      _mayFireState && !_hasPendingEvent && _hasSubscribers;
+
+
   /** Whether there is currently a subscriber on this [Stream]. */
   bool get _hasSubscribers => _subscriber != null;
 
   // -------------------------------------------------------------------
   // Internal implementation.
 
   /**
    * Create the new subscription object.
    */
   _StreamSubscriptionImpl<T> _createSubscription(
       void onData(T data),
       void onError(AsyncError error),
       void onDone(),
       bool unsubscribeOnError) {
     return new _StreamSubscriptionImpl<T>(
         this, onData, onError, onDone, unsubscribeOnError);
   }
 
   void _addListener(_StreamListener subscription) {
     if (_hasSubscribers) {
       throw new StateError("Stream already has subscriber.");
     }
     _subscriber = subscription;
     subscription._setSubscribed(0);
-    _onSubscriptionStateChange();
+    _callOnSubscriptionStateChange();
     if (_hasPendingEvent) {
       _schedulePendingEvents();
     }
   }
 
   /**
    * Handle a cancel requested from a [_StreamSubscriptionImpl].
    *
    * This method is called from [_StreamSubscriptionImpl.cancel].
    *
    * If an event is currently firing, the cancel is delayed
    * until after the subscriber has received the event.
    */
   void _cancel(_StreamListener subscriber) {
     assert(identical(subscriber._source, this));
     // We allow unsubscribing the currently firing subscription during
     // the event firing, because it is indistinguishable from delaying it since
     // that event has already received the event.
     if (!identical(_subscriber, subscriber)) {
       // You may unsubscribe more than once, only the first one counts.
       return;
     }
     _subscriber = null;
     // Unsubscribing a paused subscription also cancels its pauses.
     int subscriptionPauseCount = subscriber._setUnsubscribed();
     _updatePauseCount(-subscriptionPauseCount);
     if (!_isFiring) {
-      if (subscriptionPauseCount > 0) {
-        _onPauseStateChange();
-      }
-      _onSubscriptionStateChange();
+      _callOnSubscriptionStateChange();
     }
   }
 
   void _forEachSubscriber(
       void action(_StreamListener<T> subscription)) {
     assert(!_isPaused);
     _StreamListener subscription = _subscriber;
     assert(subscription != null);
     _startFiring();
     action(subscription);
@@ skipping 98 matching lines @@
       }
     }
     _endFiring();
   }
 
   void _addListener(_StreamListener listener) {
     listener._setSubscribed(_currentEventIdBit);
     bool firstSubscriber = !_hasSubscribers;
     _InternalLinkList.add(this, listener);
     if (firstSubscriber) {
-      _onSubscriptionStateChange();
+      _callOnSubscriptionStateChange();
     }
   }
 
   /**
    * Handle a cancel requested from a [_StreamListener].
    *
    * This method is called from [_StreamListener.cancel].
    *
    * If an event is currently firing, the cancel is delayed
    * until after the subscribers have received the event.
@@ skipping 12 matching lines @@
         // The listener has been notified of the event (or don't need to,
         // if it's still pending subscription) so it's safe to remove it.
         _removeListener(listener);
       }
       // Pause and subscription state changes are reported when we end
       // firing.
     } else {
       bool wasPaused = _isPaused;
       _removeListener(listener);
       if (wasPaused != _isPaused) _onPauseStateChange();
-      if (!_hasSubscribers) _onSubscriptionStateChange();
+      if (!_hasSubscribers) _callOnSubscriptionStateChange();
     }
   }
 
   /**
    * Removes a listener from this stream and cancels its pauses.
    *
    * This is a low-level action that doesn't call [_onSubscriptionStateChange].
-   * or [_onPauseStateChange].
+   * or [_callOnPauseStateChange].
    */
   void _removeListener(_StreamListener listener) {
     int pauseCount = listener._setUnsubscribed();
     _updatePauseCount(-pauseCount);
     _InternalLinkList.remove(listener);
   }
 }
 
 
 /** Stream that generates its own events. */
@@ skipping 469 matching lines @@
     _pauseCount = 0;
   }
 }
 
 class _SingleStreamMultiplexer<T> extends _MultiStreamImpl<T> {
   final _SingleStreamImpl<T> _source;
   StreamSubscription<T> _subscription;
 
   _SingleStreamMultiplexer(this._source);
 
-  void _onPauseStateChange() {
+  void _callOnPauseStateChange() {
     if (_isPaused) {
       if (_subscription != null) {
         _subscription.pause();
       }
     } else {
       if (_subscription != null) {
         _subscription.resume();
       }
     }
   }
 
   /**
     * Subscribe or unsubscribe on [_source] depending on whether
     * [_stream] has subscribers.
     */
   void _onSubscriptionStateChange() {
     if (_hasSubscribers) {
       assert(_subscription == null);
       _subscription = _source.listen(this._add,
                                      onError: this._signalError,
                                      onDone: this._close);
     } else {
       // TODO(lrn): Check why this can happen.
       if (_subscription == null) return;
       _subscription.cancel();
       _subscription = null;
     }
   }
 }