Avoid the _onSubscriptionStateChange being called twice in some cases.

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;
@@ skipping 186 matching lines @@
     _state |= _STREAM_CLOSED;
   }
 
   void _setComplete() {
     assert(_isClosed);
     _state = _state |_STREAM_COMPLETE;
   }
 
   void _startFiring() {
     assert(!_isFiring);
+    assert(_hasSubscribers);
+    assert(!_isPaused);
     // 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();
   }
 
   /**
    * 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) {
+    if (!wasPaused && !_isFiring) {
       _onPauseStateChange();
     }
   }
 
   /** 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) {
-      _onPauseStateChange();
+      if (!_isFiring) _onPauseStateChange();
       if (_hasPendingEvent) {
         // If we can fire events now, fire any pending events right away.
         if (fromEvent && !_isFiring) {
           _handlePendingEvents();
         } else {
-          _pendingEvents.schedule(this);
+          _schedulePendingEvents();
         }
       }
     }
   }
 
+  /** Schedule pending events to be executed. */
+  void _schedulePendingEvents() {
+    assert(_hasPendingEvent);
+    _pendingEvents.schedule(this);
+  }
+
   /** Create a subscription object. Called by [subcribe]. */
   _StreamSubscriptionImpl<T> _createSubscription(
       void onData(T data),
       void onError(AsyncError error),
       void onDone(),
       bool unsubscribeOnError);
 
   /**
    * Adds a listener to this stream.
    */
@@ skipping 103 matching lines @@
       _cancel(subscriber);
       try {
         subscriber._sendDone();
       } on AsyncError catch (e) {
         e.throwDelayed();
       } catch (e, s) {
         new AsyncError(e, s).throwDelayed();
       }
     });
     assert(!_hasSubscribers);
-    _onSubscriptionStateChange();
   }
 }
 
 // -------------------------------------------------------------------
 // Default implementation of a stream with a single subscriber.
 // -------------------------------------------------------------------
 /**
  * Default implementation of stream capable of sending events to one subscriber.
  *
  * Any class needing to implement [Stream] can either directly extend this
@@ skipping 49 matching lines @@
   }
 
   void _addListener(_StreamListener subscription) {
     if (_hasSubscribers) {
       throw new StateError("Stream already has subscriber.");
     }
     _subscriber = subscription;
     subscription._setSubscribed(0);
     _onSubscriptionStateChange();
     if (_hasPendingEvent) {
-      new Timer(0, (_) {
-        _handlePendingEvents();
-      });
+      _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;
     int timesPaused = subscriber._setUnsubscribed();

[floitsch, 2013/01/14 16:31:24]

Add comment: Unsubscribing a paused subscriber cancels the pauses. (Or similar).

s/timesPaused/subscriptionPauseCount

[Lasse Reichstein Nielsen, 2013/01/15 08:52:54]

Done.

     _updatePauseCount(-timesPaused);
-    if (timesPaused > 0) {
-      _onPauseStateChange();
+    if (!_isFiring) {
+      if (timesPaused > 0) {
+        _onPauseStateChange();
+      }
+      _onSubscriptionStateChange();
     }
-    _onSubscriptionStateChange();
   }
 
   void _forEachSubscriber(
       void action(_StreamListener<T> subscription)) {
+    assert(!_isPaused);
     _StreamListener subscription = _subscriber;
     assert(subscription != null);
     _startFiring();
     action(subscription);
     _endFiring();
   }
 }
 
 // -------------------------------------------------------------------
 // Default implementation of a stream with subscribers.
@@ skipping 83 matching lines @@
         action(current);
         // Marks as having received the event.
         current._toggleEventReceived();
       }
       cursor = current._nextLink;
       if (current._isPendingUnsubscribe) {
         _removeListener(current);
       }
     }
     _endFiring();
-    if (_isPaused) _onPauseStateChange();
-    if (!_hasSubscribers) _onSubscriptionStateChange();
   }
 
   void _addListener(_StreamListener listener) {
     listener._setSubscribed(_currentEventIdBit);
     bool firstSubscriber = !_hasSubscribers;
     _InternalLinkList.add(this, listener);
     if (firstSubscriber) {
       _onSubscriptionStateChange();
     }
   }
@@ skipping 19 matching lines @@
       } else {
         // 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 (!identical(wasPaused, _isPaused)) _onPauseStateChange();

[floitsch, 2013/01/14 16:31:24]

why?

[Lasse Reichstein Nielsen, 2013/01/15 08:52:54]

Probably overoptimizing :)
Reverted.

       if (!_hasSubscribers) _onSubscriptionStateChange();
     }
   }
 
   /**
    * Removes a listener from this stream and cancels its pauses.
    *
    * This is a low-level action that doesn't call [_onSubscriptionStateChange].
    * or [_onPauseStateChange].
    */
   void _removeListener(_StreamListener listener) {
     int pauseCount = listener._setUnsubscribed();
     _updatePauseCount(-pauseCount);
     _InternalLinkList.remove(listener);
   }
 }
 
 
 /** Abstract superclass for streams that generate their own events. */
 abstract class _GeneratedSingleStreamImpl<T> extends _SingleStreamImpl<T> {
   bool _isHandlingPendingEvents = false;
   bool get _hasPendingEvent => !_isClosed;
 
+  void _schedulePendingEvents() {
+    if (_pendingEvents != null) {
+      _pendingEvents.schedule(this);
+    } else {
+      // In the case where there only pending events are generated ones.

[floitsch, 2013/01/14 16:31:24]

Don't understand comment.

[Lasse Reichstein Nielsen, 2013/01/15 08:52:54]

It's commenting that in this particular class, it's possible to have pending
events even if there is no _pendingEvents object, which is why we can't use the
branch before.
Comment updated.

I think I'll change how generated events work, and instead generate them inside
the pending events object, so the rest of the object doesn't need changing.

[floitsch, 2013/01/15 14:40:19]

Did you commit? The comment still looks the same.

+      new Timer(0, (_) { _handlePendingEvents(); });
+    }
+  }
+
   /**
    * Generate one (or possibly more) new events.
    *
    * The events should be added to the stream using [_add], [_signalError] and
    * [_close].
    */
   void _generateNextEvent();
 
   void _handlePendingEvents() {
     // Avoid reentry from _add/_signalError/_close potentially called
@@ skipping 418 matching lines @@
     if (_isComplete) {
       throw new StateError("Subscription has been canceled.");
     }
     if (_timer != null) {
       _timer.cancel();
       _timer = null;
     }
     _pauseCount = 0;
   }
 }