Change and structure how Stream implementations do 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.
 
+// Completion state of the stream.
 /// 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;
+
+// The activity state of the stream: What is it currently doing.
 /// 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 pause state of the stream.
+/// Bit set when resuming with pending events. Cleared after all pending events
+/// have been transmitted. Means that the controller still considers the
+/// stream paused, even if the listener doesn't.
+const int _STREAM_PENDING_RESUME = 32;
 /// The count of times a stream has paused is stored in the
 /// state, shifted by this amount.
-const int _STREAM_PAUSE_COUNT_SHIFT = 5;
+const int _STREAM_PAUSE_COUNT_SHIFT = 6;
 
 // 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;
+
 /// Bit that contains the last sent event's "id bit".
 const int _LISTENER_EVENT_ID = 4;
 const int _LISTENER_EVENT_ID_SHIFT = 2;
+
 /// The count of times a listener has paused is stored in the
 /// state, shifted by this amount.
 const int _LISTENER_PAUSE_COUNT_SHIFT = 3;
 
 
 // -------------------------------------------------------------------
 // Common base class for single and multi-subscription streams.
 // -------------------------------------------------------------------
 abstract class _StreamImpl<T> extends Stream<T> {
   /** Current state of the stream. */
@@ skipping 33 matching lines @@
   // StreamSink interface-like methods for sending events into the stream.
   // It's the responsibility of the caller to ensure that the stream is not
   // paused when adding events. If the stream is paused, the events will be
   // queued, but it's better to not send events at all.
 
   /**
    * Send or queue a data event.
    */
   void _add(T value) {
     if (_isClosed) throw new StateError("Sending on closed stream");
-    if (!_canFireEvent) {
+    if (!_mayFireState) {
+      // Not the time to send events.
       _addPendingEvent(new _DelayedData<T>(value));
       return;
     }
-    _sendData(value);
+    if (_hasPendingEvent) {
+      _addPendingEvent(new _DelayedData<T>(value));
+    } else {
+      _sendData(value);
+    }
     _handlePendingEvents();
   }
 
   /**
    * Send or enqueue an error event.
    *
    * If a subscription has requested to be unsubscribed on errors,
    * it will be unsubscribed after receiving this event.
    */
   void _signalError(AsyncError error) {
     if (_isClosed) throw new StateError("Sending on closed stream");
-    if (!_canFireEvent) {
+    if (!_mayFireState) {
+      // Not the time to send events.
       _addPendingEvent(new _DelayedError(error));
       return;
     }
-    _sendError(error);
+    if (_hasPendingEvent) {
+      _addPendingEvent(new _DelayedError(error));
+    } else {
+      _sendError(error);
+    }
     _handlePendingEvents();
   }
 
   /**
    * Send or enqueue a "done" message.
    *
    * The "done" message should be sent at most once by a stream, and it
    * should be the last message sent.
    */
   void _close() {
     if (_isClosed) return;
     _state |= _STREAM_CLOSED;
-    if (!_canFireEvent) {
-      // You can't enqueue an event after the Done, so make it const.
+    if (!_mayFireState) {
+      // Not the time to send events.
       _addPendingEvent(const _DelayedDone());
       return;
     }
-    _sendDone();
-    assert(!_hasPendingEvent);
+    if (_hasPendingEvent) {
+      _addPendingEvent(new _DelayedDone());
+      _handlePendingEvents();
+    } else {
+      _sendDone();
+      assert(_isComplete);
+      assert(!_hasPendingEvent);
+    }
   }
 
   // -------------------------------------------------------------------
   // Internal implementation.
 
   // State prediates.

[floitsch, 2013/03/01 21:52:30]

predicates

[Lasse Reichstein Nielsen, 2013/03/04 11:53:02]

Done.

 
+  // Lifecycle state.
+  /** Whether the stream is in the default, open, state for events. */
+  bool get _isOpen => (_state & (_STREAM_CLOSED | _STREAM_COMPLETE)) == 0;
+
   /** Whether the stream has been closed (a done event requested). */
   bool get _isClosed => (_state & _STREAM_CLOSED) != 0;
 
   /** Whether the stream is completed. */
   bool get _isComplete => (_state & _STREAM_COMPLETE) != 0;
 
+  // Pause state.
+
   /** Whether one or more active subscribers have requested a pause. */
   bool get _isPaused => _state >= (1 << _STREAM_PAUSE_COUNT_SHIFT);
 
+  /** How many times the stream has been paused. */
+  int get _pauseCount => _state >> _STREAM_PAUSE_COUNT_SHIFT;
+
+  /**
+   * Whether a controller thinks the stream is paused.
+   *
+   * When this changes, a pause-state change callback is performed.
+   *
+   * It may differ from [_isPaused] if there are pending events
+   * in the queue when the listeners resume. The controller won't
+   * be informed until all queued events have been fired.
+   */
+  bool get _isControllerPaused => _state >= (_STREAM_PENDING_RESUME);
+
+  /** Whether we have a pending resume scheduled. */
+  bool get _hasPendingResume => (_state & _STREAM_PENDING_RESUME) != 0;
+
+
+  // Action state. If the stream makes a call-out to external code,
+  // this state tracks it and avoids reentrancy problems.
+
+  /** Whether the stream is not currently firing or calling a callback. */
+  bool get _isInactive => (_state & (_STREAM_CALLBACK | _STREAM_FIRING)) == 0;
+
   /** Whether we are currently executing a state-chance callback. */
   bool get _isInCallback => (_state & _STREAM_CALLBACK) != 0;
 
+  /** Whether we are currently firing an event. */
+  bool get _isFiring => (_state & _STREAM_FIRING) != 0;
+
   /** 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;
+  /**
+   * The bit representing the current or last event fired.
+   *
+   * This bit matches a bit on listeners that have received the corresponding
+   * event. It is toggled for each new event being fired.
+   */
+  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 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.
+    // The state allows firing unless:
+    // - it's currently firing
+    // - it's currently in a callback
+    // - it's paused
     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 => _mayFireState && !_hasPendingEvent;
-
   // State modification.
 
   /** Record an increases in the number of times the listener has paused. */
   void _incrementPauseCount(_StreamListener<T> listener) {
     listener._incrementPauseCount();
+    _state &= ~_STREAM_PENDING_RESUME;
     _updatePauseCount(1);
   }
 
   /** Record a decrease in the number of times the listener has paused. */
   void _decrementPauseCount(_StreamListener<T> listener) {
     assert(_isPaused);
     listener._decrementPauseCount();
     _updatePauseCount(-1);
   }
 
@@ skipping 17 matching lines @@
     _state |= _STREAM_CLOSED;
   }
 
   void _setComplete() {
     assert(_isClosed);
     _state = _state |_STREAM_COMPLETE;
   }
 
   void _startFiring() {
     assert(!_isFiring);
+    assert(!_isInCallback);
     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() {
+  void _endFiring(bool wasControllerPaused) {
     assert(_isFiring);
     _state ^= _STREAM_FIRING;
-    if (!_hasSubscribers) {
-      _callOnSubscriptionStateChange();
-    } else if (_isPaused) {
-      _callOnPauseStateChange();
-    }
+    // Had listeners, or we wouldn't have fired.
+    _checkCallbacks(true, wasControllerPaused);
   }
 
   /**
    * 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 wasControllerPaused = _isControllerPaused;

[floitsch, 2013/03/01 21:52:30]

what about calling this "wasInputPaused" ?

[Lasse Reichstein Nielsen, 2013/03/04 11:53:02]

The real name should be "wasLastPauseStateChangeCallbackAPause".
The controller itself isn't paused, it can do whatever it wants, but it thinks
the stream is paused (even though the stream might already be sending pending
out events).
But "Input" is less specific, I'll change it to that.

     bool wasPaused = _isPaused;
     _incrementPauseCount(listener);
     if (resumeSignal != null) {
       resumeSignal.whenComplete(() { this._resume(listener, true); });
     }
-    if (!wasPaused && !_isFiring) {
-      _callOnPauseStateChange();
+    if (!wasPaused && _hasPendingEvent && _pendingEvents.isScheduled) {
+      _pendingEvents.cancelSchedule();
+    }
+    if (_isInactive && !wasControllerPaused) {
+      _checkCallbacks(true, false);
+      if (!_isPaused && _hasPendingEvent) {
+        _schedulePendingEvents();
+      }
     }
   }
 
   /** 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) _callOnPauseStateChange();
       if (_hasPendingEvent) {
+        _state |= _STREAM_PENDING_RESUME;
+        // Controller's pause state haven't changed.

[floitsch, 2013/03/01 21:52:30]

hasn't

         // If we can fire events now, fire any pending events right away.
-        if (fromEvent && !_isFiring) {
-          _handlePendingEvents();
-        } else {
-          _schedulePendingEvents();
+        if (_isInactive) {
+          if (fromEvent) {
+            _handlePendingEvents();
+          } else {
+            _schedulePendingEvents();
+          }
+        }
+      } else if (_isInactive) {
+        _checkCallbacks(true, true);
+        if (!_isPaused && _hasPendingEvent) {
+          if (fromEvent) {
+            _handlePendingEvents();
+          } else {
+            _schedulePendingEvents();
+          }
         }
       }
     }
   }
 
   /** Schedule pending events to be executed. */
   void _schedulePendingEvents() {
     assert(_hasPendingEvent);
     _pendingEvents.schedule(this);
   }
@@ skipping 30 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;
-    if (!_isInCallback) {
-      _state |= _STREAM_CALLBACK;
-      _onPauseStateChange();
-      _state ^= _STREAM_CALLBACK;
-    } else {
-      _onPauseStateChange();
+  /**
+   * Checks whether the subscription/pause state has changed.
+   *
+   * Calls the appropriate callback if the state has changed from the
+   * provided one. Repeats calling callbacks as long as the call changes
+   * the state.
+   */
+  void _checkCallbacks(bool hadSubscribers, bool wasPaused) {
+    assert(!_isFiring);
+    // Will be handled after the current callback.
+    if (_isInCallback) return;
+    if (_hasPendingResume && !_hasPendingEvent) {
+      _state ^= _STREAM_PENDING_RESUME;
+    }
+    _state |= _STREAM_CALLBACK;
+    while (true) {
+      bool hasSubscribers = _hasSubscribers;
+      bool isPaused = _isControllerPaused;
+      if (hadSubscribers != hasSubscribers) {
+        _onSubscriptionStateChange();
+      } else if (isPaused != wasPaused) {
+        _onPauseStateChange();
+      } else {
+        _state ^= _STREAM_CALLBACK;
+        return;
+      }
+      wasPaused = isPaused;
+      hadSubscribers = hasSubscribers;
     }
   }
 
-  /** Calls [_onSubscriptionStateChange] while setting callback bit. */
-  void _callOnSubscriptionStateChange() {
-    if (!_isInCallback) {
-      _state |= _STREAM_CALLBACK;
-      _onSubscriptionStateChange();
-      _state ^= _STREAM_CALLBACK;
-    } else {
-      _onSubscriptionStateChange();
-    }
-  }
-
   /**
    * 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.
    *
    * Read [hasSubscribers] to see what the new state is.
    */
   void _onSubscriptionStateChange() {}
 
-  /** Add a pending event at the end of the pending event queue. */
+  /**
+   * Add a pending event at the end of the pending event queue.
+   *
+   * Only called from [_add], [_signalError] and [_close], so in some
+   * cases events may be fired from here.

[floitsch, 2013/03/01 21:52:30]

That sounds bad. adding a pending event should not trigger fire it.

_add, _signalError and _close already call _handlePendingEvents() if they are
allowed to fire.

[Lasse Reichstein Nielsen, 2013/03/04 11:53:02]

True, it is unnecessary here.

+   */
   void _addPendingEvent(_DelayedEvent event) {
     if (_pendingEvents == null) _pendingEvents = new _StreamImplEvents();
     _StreamImplEvents events = _pendingEvents;
     events.add(event);
+    if (_isPaused || _isFiring) return;
+    if (_isInCallback) {
+      _schedulePendingEvents();
+      return;
+    }
+    // Event only delayed because of other events in queue, so just fire them
+    // now.
+    _handlePendingEvents();
   }
 
-  /** Fire any pending events until the pending event queue. */
+  /** Fire any pending events until the pending event queue is empty. */
   void _handlePendingEvents() {
+    assert(_isInactive);
+    if (!_hasPendingEvent) return;
     _PendingEvents events = _pendingEvents;
-    if (events == null) return;
-    while (!events.isEmpty && !_isPaused) {
+    do {
+      if (_isPaused) return;
+      if (events.isScheduled) events.cancelSchedule();
       events.handleNext(this);
-    }
+    } while (!events.isEmpty);
   }
 
   /**
    * Send a data event directly to each subscriber.
    */
   _sendData(T value) {
     assert(!_isPaused);
     assert(!_isComplete);
+    if (!_hasSubscribers) return;
     _forEachSubscriber((subscriber) {
       try {
         subscriber._sendData(value);
       } on AsyncError catch (e) {
         e.throwDelayed();
       } catch (e, s) {
         new AsyncError(e, s).throwDelayed();
       }
     });
   }
 
   /**
    * Sends an error event directly to each subscriber.
    */
   void _sendError(AsyncError error) {
     assert(!_isPaused);
     assert(!_isComplete);
+    if (!_hasSubscribers) return;
     _forEachSubscriber((subscriber) {
       try {
         subscriber._sendError(error);
       } on AsyncError catch (e) {
         e.throwDelayed();
       } catch (e, s) {
         new AsyncError.withCause(e, s, error).throwDelayed();
       }
     });
   }
@@ skipping 40 matching lines @@
  * some methods made public.
  *
  * The user interface of [_SingleStreamImpl] are the following methods:
  * * [_add]: Add a data event to the stream.
  * * [_signalError]: Add an error event to the stream.
  * * [_close]: Request to close the stream.
  * * [_onSubscriberStateChange]: Called when receiving the first subscriber or
  *                               when losing the last subscriber.
  * * [_onPauseStateChange]: Called when entering or leaving paused mode.
  * * [_hasSubscribers]: Test whether there are currently any subscribers.
- * * [_isPaused]: Test whether the stream is currently paused.
+ * * [_isControllerPaused]: Test whether the stream is currently paused.

[floitsch, 2013/03/01 21:52:30]

We still have the "_isPaused" too. Is that one not considered "public" anymore?

[Lasse Reichstein Nielsen, 2013/03/04 11:53:02]

True, it's not considered "public".
Only use it internally if you want to know if sending events is blocked.

  * 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;
 
-  // 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;
-
-
-  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]. */
   bool get _hasSubscribers => _subscriber != null;
 
   // -------------------------------------------------------------------
   // Internal implementation.
 
+  _SingleStreamImpl() {
+    // Start out paused.
+    _updatePauseCount(1);
+  }
+
   /**
    * 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) {
+    assert(!_isComplete);
     if (_hasSubscribers) {
       throw new StateError("Stream already has subscriber.");
     }
+    assert(_pauseCount == 1);
+    _updatePauseCount(-1);
     _subscriber = subscription;
     subscription._setSubscribed(0);
-    _callOnSubscriptionStateChange();
-    if (_hasPendingEvent) {
-      _schedulePendingEvents();
+    if (_isInactive) {
+      _checkCallbacks(false, true);
+      if (!_isPaused && _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) {
-      _callOnSubscriptionStateChange();
+    int resumeCount = subscriber._setUnsubscribed();
+    // Keep being paused while there is no subscriber and the stream is not

[floitsch, 2013/03/01 21:52:30]

I don't understand this:
why do we pause a Stream without a subscriber?
when do we call the subscriptionStateChange?

[Lasse Reichstein Nielsen, 2013/03/04 11:53:02]

This is how single-subscription streams have worked so far.
Previously it was hacked in by changing the _isPaused predicate. I'm making it
explicit.

The subscriber callback is called by _checkCallbacks.

+    // complete.
+    _updatePauseCount(_isComplete ? -resumeCount : -resumeCount + 1);
+    if (_isInactive) {
+      _checkCallbacks(true, resumeCount > 0);
+      if (!_isPaused && _hasPendingEvent) {
+        _schedulePendingEvents();
+      }
     }
   }
 
   void _forEachSubscriber(
       void action(_StreamListener<T> subscription)) {
     assert(!_isPaused);
+    bool wasControllerPaused = _isControllerPaused;
     _StreamListener subscription = _subscriber;
     assert(subscription != null);
     _startFiring();
     action(subscription);
-    _endFiring();
+    _endFiring(wasControllerPaused);
   }
 }
 
 // -------------------------------------------------------------------
 // Default implementation of a stream with subscribers.
 // -------------------------------------------------------------------
 
 /**
  * Default implementation of stream capable of sending events to subscribers.
  *
@@ skipping 65 matching lines @@
    *
    * 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(_StreamListener<T> subscription)) {
     assert(!_isFiring);
     if (!_hasSubscribers) return;
+    bool wasControllerPaused = _isControllerPaused;
     _startFiring();
     _InternalLink cursor = this._nextLink;
     while (!identical(cursor, this)) {
       _StreamListener<T> current = cursor;
       if (current._needsEvent(_currentEventIdBit)) {
         action(current);
         // Marks as having received the event.
         current._toggleEventReceived();
       }
       cursor = current._nextLink;
       if (current._isPendingUnsubscribe) {
         _removeListener(current);
       }
     }
-    _endFiring();
+    _endFiring(wasControllerPaused);
   }
 
   void _addListener(_StreamListener listener) {
     listener._setSubscribed(_currentEventIdBit);
-    bool firstSubscriber = !_hasSubscribers;
+    bool hadSubscribers = _hasSubscribers;
     _InternalLinkList.add(this, listener);
-    if (firstSubscriber) {
-      _callOnSubscriptionStateChange();
+    if (!hadSubscribers && _isInactive) {
+      _checkCallbacks(false, false);
+      if (!_isPaused && _hasPendingEvent) {
+        _schedulePendingEvents();
+      }
     }
   }
 
   /**
    * 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.
    */
   void _cancel(_StreamListener listener) {
     assert(identical(listener._source, this));
     if (_InternalLink.isUnlinked(listener)) {
       // You may unsubscribe more than once, only the first one counts.
       return;
     }
     if (_isFiring) {
       if (listener._needsEvent(_currentEventIdBit)) {
         assert(listener._isSubscribed);
-        listener._setPendingUnsubscribe();
+        listener._setPendingUnsubscribe(_currentEventIdBit);
       } 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;
+      bool wasControllerPaused = _isControllerPaused;
       _removeListener(listener);
-      if (wasPaused != _isPaused) _onPauseStateChange();
-      if (!_hasSubscribers) _callOnSubscriptionStateChange();
+      if (_isInactive) {
+        _checkCallbacks(true, wasControllerPaused);
+        if (!_isPaused && _hasPendingEvent) {
+          _schedulePendingEvents();
+        }
+      }
     }
   }
 
   /**
    * Removes a listener from this stream and cancels its pauses.
    *
    * This is a low-level action that doesn't call [_onSubscriptionStateChange].
    * or [_callOnPauseStateChange].
    */
   void _removeListener(_StreamListener listener) {
     int pauseCount = listener._setUnsubscribed();
-    _updatePauseCount(-pauseCount);
     _InternalLinkList.remove(listener);
+    if (pauseCount > 0) {
+      _updatePauseCount(-pauseCount);
+      if (!_isPaused && _hasPendingEvent) {
+        _state |= _STREAM_PENDING_RESUME;
+      }
+    }
   }
 }
 
 
 /** Stream that generates its own events. */
 class _GeneratedSingleStreamImpl<T> extends _SingleStreamImpl<T> {
   /**
    * Initializes the stream to have only the events provided by [events].
    *
    * A [_PendingEvents] implementation provides events that are handled
@@ skipping 304 matching lines @@
   /// the event.
   void _toggleEventReceived() {
     _state ^= _LISTENER_EVENT_ID;
   }
 
   void _setSubscribed(int eventIdBit) {
     assert(eventIdBit == 0 || eventIdBit == 1);
     _state = _LISTENER_SUBSCRIBED | (eventIdBit << _LISTENER_EVENT_ID_SHIFT);
   }
 
-  void _setPendingUnsubscribe() {
+  void _setPendingUnsubscribe(int currentEventIdBit) {
     assert(_isSubscribed);
-    _state |= _LISTENER_PENDING_UNSUBSCRIBE;
+    // Sets the pending unsubscribe, and ensures that the listener
+    // won't get the current event.
+    _state |= _LISTENER_PENDING_UNSUBSCRIBE | _LISTENER_EVENT_ID;
+    _state ^= (1 ^ currentEventIdBit) << _LISTENER_EVENT_ID_SHIFT;
+    assert(!_needsEvent(currentEventIdBit));
   }
 
   /**
    * Marks the listener as unsubscibed.
    *
    * Returns the number of unresumed pauses for the listener.
    */
   int _setUnsubscribed() {
     assert(_isSubscribed);
     int timesPaused = _state >> _LISTENER_PAUSE_COUNT_SHIFT;
@@ skipping 61 matching lines @@
 
   void add(_DelayedEvent event) {
     if (lastPendingEvent == null) {
       firstPendingEvent = lastPendingEvent = event;
     } else {
       lastPendingEvent = lastPendingEvent.next = event;
     }
   }
 
   void handleNext(_StreamImpl stream) {
-    if (isScheduled) cancelSchedule();
+    assert(!isScheduled);
     _DelayedEvent event = firstPendingEvent;
     firstPendingEvent = event.next;
     if (firstPendingEvent == null) {
       lastPendingEvent = null;
     }
     event.perform(stream);
   }
 }
 
 
@@ skipping 20 matching lines @@
   void onError(void handleError(AsyncError error)) {}
 
   void onDone(void handleDone()) {
     _handler = handleDone;
   }
 
   void pause([Future signal]) {
     if (_isComplete) {
       throw new StateError("Subscription has been canceled.");
     }
-    if (_timer != null) _timer.cancel();
+    if (_timer != null) {
+      _timer.cancel();
+      _timer = null;
+    }
     _pauseCount++;
+    if (signal != null) signal.whenComplete(resume);
   }
 
   void resume() {
     if (_isComplete) {
       throw new StateError("Subscription has been canceled.");
     }
     if (_pauseCount == 0) return;
     _pauseCount--;
     if (_pauseCount == 0) {
       _delayDone();
@@ skipping 43 matching lines @@
                                      onError: this._signalError,
                                      onDone: this._close);
     } else {
       // TODO(lrn): Check why this can happen.
       if (_subscription == null) return;
       _subscription.cancel();
       _subscription = null;
     }
   }
 }