Change hasSubscribers to hasListener.

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.
@@ skipping 208 matching lines @@
   /**
    * 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;
+  bool get _hasListener;
 
 
   /** Whether the state bits allow firing. */
   bool get _mayFireState {
     // The state allows firing unless:
     // - it's currently firing
     // - it's currently in a callback
     // - it's paused
     const int mask =
         _STREAM_FIRING |
@@ skipping 39 matching lines @@
   }
 
   void _setComplete() {
     assert(_isClosed);
     _state = _state |_STREAM_COMPLETE;
   }
 
   void _startFiring() {
     assert(!_isFiring);
     assert(!_isInCallback);
-    assert(_hasSubscribers);
+    assert(_hasListener);
     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(bool wasInputPaused) {
     assert(_isFiring);
@@ skipping 110 matching lines @@
    */
   void _forEachSubscriber(void action(_StreamSubscriptionImpl<T> subscription));
 
   /**
    * 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) {
+  void _checkCallbacks(bool hadListener, 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 hasListener = _hasListener;
       bool isPaused = _isInputPaused;
-      if (hadSubscribers != hasSubscribers) {
+      if (hadListener != hasListener) {
         _onSubscriptionStateChange();
       } else if (isPaused != wasPaused) {
         _onPauseStateChange();
       } else {
         _state ^= _STREAM_CALLBACK;
         return;
       }
       wasPaused = isPaused;
-      hadSubscribers = hasSubscribers;
+      hadListener = hasListener;
     }
   }
 
   /**
    * 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.
+   * Read [hasListener] to see what the new state is.
    */
   void _onSubscriptionStateChange() {}
 
   /**
    * 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();
@@ skipping 17 matching lines @@
       events.handleNext(this);
     } while (!events.isEmpty);
   }
 
   /**
    * Send a data event directly to each subscriber.
    */
   _sendData(T value) {
     assert(!_isPaused);
     assert(!_isComplete);
-    if (!_hasSubscribers) return;
+    if (!_hasListener) 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;
+    if (!_hasListener) return;
     _forEachSubscriber((subscriber) {
       try {
         subscriber._sendError(error);
       } on AsyncError catch (e) {
         e.throwDelayed();
       } catch (e, s) {
         new AsyncError.withCause(e, s, error).throwDelayed();
       }
     });
   }
 
   /**
    * Sends the "done" message directly to each subscriber.
    * This automatically stops further subscription and
    * unsubscribes all subscribers.
    */
   void _sendDone() {
     assert(!_isPaused);
     assert(_isClosed);
     _setComplete();
-    if (!_hasSubscribers) return;
+    if (!_hasListener) return;
     _forEachSubscriber((subscriber) {
       _cancel(subscriber);
       try {
         subscriber._sendDone();
       } on AsyncError catch (e) {
         e.throwDelayed();
       } catch (e, s) {
         new AsyncError(e, s).throwDelayed();
       }
     });
-    assert(!_hasSubscribers);
+    assert(!_hasListener);
   }
 }
 
 // -------------------------------------------------------------------
 // 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
  * class, or extend [Stream] and delegate the subscribe method to an instance
  * of this class.
  *
  * The only public methods are those of [Stream], so instances of
  * [_SingleStreamImpl] can be returned directly as a [Stream] without exposing
  * internal functionality.
  *
  * The [StreamController] is a public facing version of this class, with
  * some methods made public.
  *
  * The user interface of [_SingleStreamImpl] are the following methods:
  * * [_add]: Add a data event to the stream.
  * * [_addError]: 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.
+ * * [_hasListener]: Test whether there are currently any subscribers.
  * * [_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;
 
   /** Whether there is currently a subscriber on this [Stream]. */
-  bool get _hasSubscribers => _subscriber != null;
+  bool get _hasListener => _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) {
+    if (_hasListener) {
       throw new StateError("Stream already has subscriber.");
     }
     assert(_pauseCount == 1);
     _updatePauseCount(-1);
     _subscriber = subscription;
     subscription._setSubscribed(0);
     if (_isInactive) {
       _checkCallbacks(false, true);
       if (!_isPaused && _hasPendingEvent) {
         _schedulePendingEvents();
@@ skipping 59 matching lines @@
  * The [StreamController] is a public facing version of this class, with
  * some methods made public.
  *
  * The user interface of [_MultiStreamImpl] are the following methods:
  * * [_add]: Add a data event to the stream.
  * * [_addError]: Add an error event to the stream.
  * * [_close]: Request to close the stream.
  * * [_onSubscriptionStateChange]: 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.
+ * * [_hasListener]: 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 _MultiStreamImpl<T> extends _StreamImpl<T>
                           implements _InternalLinkList {
   // Link list implementation (mixin when possible).
   _InternalLink _nextLink;
   _InternalLink _previousLink;
 
   _MultiStreamImpl() {
     _nextLink = _previousLink = this;
   }
 
   bool get isBroadcast => true;
 
   Stream<T> asBroadcastStream() => this;
 
   // ------------------------------------------------------------------
   // Helper functions that can be overridden in subclasses.
 
   /** Whether there are currently any subscribers on this [Stream]. */
-  bool get _hasSubscribers => !_InternalLinkList.isEmpty(this);
+  bool get _hasListener => !_InternalLinkList.isEmpty(this);
 
   /**
    * Create the new subscription object.
    */
   _StreamListener<T> _createSubscription(
       void onData(T data),
       void onError(AsyncError error),
       void onDone(),
       bool unsubscribeOnError) {
     return new _StreamSubscriptionImpl<T>(
@@ skipping 11 matching lines @@
    * after the iteration is complete.
    *
    * 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;
+    if (!_hasListener) 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(wasInputPaused);
   }
 
   void _addListener(_StreamListener listener) {
     listener._setSubscribed(_currentEventIdBit);
-    bool hadSubscribers = _hasSubscribers;
+    bool hadListener = _hasListener;
     _InternalLinkList.add(this, listener);
-    if (!hadSubscribers && _isInactive) {
+    if (!hadListener && _isInactive) {
       _checkCallbacks(false, false);
       if (!_isPaused && _hasPendingEvent) {
         _schedulePendingEvents();
       }
     }
   }
 
   /**
    * Handle a cancel requested from a [_StreamListener].
    *
@@ skipping 542 matching lines @@
         _subscription.resume();
       }
     }
   }
 
   /**
     * Subscribe or unsubscribe on [_source] depending on whether
     * [_stream] has subscribers.
     */
   void _onSubscriptionStateChange() {
-    if (_hasSubscribers) {
+    if (_hasListener) {
       assert(_subscription == null);
       _subscription = _source.listen(this._add,
                                      onError: this._addError,
                                      onDone: this._close);
     } else {
       // TODO(lrn): Check why this can happen.
       if (_subscription == null) return;
       _subscription.cancel();
       _subscription = null;
     }
   }
 }