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.
+  // State predicates.
+
+  // 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 _isInputPaused => _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 wasInputPaused) {
     assert(_isFiring);
     _state ^= _STREAM_FIRING;
-    if (!_hasSubscribers) {
-      _callOnSubscriptionStateChange();
-    } else if (_isPaused) {
-      _callOnPauseStateChange();
-    }
+    // Had listeners, or we wouldn't have fired.
+    _checkCallbacks(true, wasInputPaused);
   }
 
   /**
    * 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 wasInputPaused = _isInputPaused;
     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 && !wasInputPaused) {
+      _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 hasn't changed.
         // 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 = _isInputPaused;
+      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.
+   *
+   * Schedules events if currently not paused and inside a callback.
+   */
   void _addPendingEvent(_DelayedEvent event) {
     if (_pendingEvents == null) _pendingEvents = new _StreamImplEvents();
     _StreamImplEvents events = _pendingEvents;
     events.add(event);
-  }
-
-  /** Fire any pending events until the pending event queue. */
-  void _handlePendingEvents() {
-    _PendingEvents events = _pendingEvents;
-    if (events == null) return;
-    while (!events.isEmpty && !_isPaused) {
-      events.handleNext(this);
+    if (_isPaused || _isFiring) return;
+    if (_isInCallback) {
+      _schedulePendingEvents();
+      return;
     }
   }
 
+  /** Fire any pending events until the pending event queue is empty. */
+  void _handlePendingEvents() {
+    assert(_isInactive);
+    if (!_hasPendingEvent) return;
+    _PendingEvents events = _pendingEvents;
+    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.
+ * * [_isInputPaused]: 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;
 
-  // 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
+    // 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 wasInputPaused = _isInputPaused;
     _StreamListener subscription = _subscriber;
     assert(subscription != null);
     _startFiring();
     action(subscription);
-    _endFiring();
+    _endFiring(wasInputPaused);
   }
 }
 
 // -------------------------------------------------------------------
 // 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 wasInputPaused = _isInputPaused;
     _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(wasInputPaused);
   }
 
   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 wasInputPaused = _isInputPaused;
       _removeListener(listener);
-      if (wasPaused != _isPaused) _onPauseStateChange();
-      if (!_hasSubscribers) _callOnSubscriptionStateChange();
+      if (_isInactive) {
+        _checkCallbacks(true, wasInputPaused);
+        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;
     }
   }
 }