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 23 matching lines @@
 class _SingleStreamImpl<T> extends _StreamImpl<T> {
   _StreamListener _subscriber = null;
 
   Stream<T> asMultiSubscriberStream() {
     return new _SingleStreamMultiplexer<T>(this);
   }
 
   bool get isSingleSubscription => true;
 
   /** Whether one or more active subscribers have requested a pause. */
-  bool get _isPaused => !_hasSubscribers || super._isPaused;
+  bool get _isPaused => (!_hasSubscribers && !_isClosed) || super._isPaused;
 
   /** 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();
     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();
-    _updatePauseCount(-timesPaused);
-    if (timesPaused > 0) {
-      _onPauseStateChange();
+    // Unsubscribing a paused subscription also cancels its pauses.
+    int subscriptionPauseCount = subscriber._setUnsubscribed();
+    _updatePauseCount(-subscriptionPauseCount);
+    if (!_isFiring) {
+      if (subscriptionPauseCount > 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 85 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 43 matching lines @@
     _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.
+      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 455 matching lines @@
                                      onError: this._signalError,
                                      onDone: this._close);
     } else {
       // TODO(lrn): Check why this can happen.
       if (_subscription == null) return;
       _subscription.cancel();
       _subscription = null;
     }
   }
 }