Change and structure how Stream implementations do callbacks.

sdk/lib/async/stream_controller.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;
 
 // -------------------------------------------------------------------
 // Controller for creating and adding events to a stream.
 // -------------------------------------------------------------------
 
 /**
  * A controller with the stream it controls.
  *
  * This controller allows sending data, error and done events on
  * its [stream].
  * This class can be used to create a simple stream that others
  * can listen on, and to push events to that stream.
  *
  * It's possible to check whether the stream is paused or not, and whether
  * it has subscribers or not, as well as getting a callback when either of
  * these change.
+ *
+ * If the stream starts or stops having listeners (first listener subscribing,

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

We could make our lives easier by making the StreamController
single-subscription.
Users could use "asBroadcast" if they really want multiple subscribers.

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

An asBroadCast single-subscription stream is not the same as a real broadcast
stream. The asBroadCast-single-stream is paused, like a single-subscription
stream, until the first subscriber arrives, where a real broadcast stream is
unpaused when it's created.

+ * last listener unsubscribing), the `onSubscriptionStateChange` callback
+ * is notified as soon as possible. That will be as soon as a currently firing
+ * event is done firing, or after another callback has finished.
+ * If the pause state has also changed, only the subscription state callback

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

Not clear: are you saying that a subscription change will hide a pause-change?

so stream.listen(...).pause() will not call the pause-change?

The way I read it, this is not the case, since both the 'listen' and the 'pause'
would be immediate, but it is not completely clear.

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

Correct, if during the same callback/event, both subscription and pause is
changed, only a subscription event is passed. At any other time, the callback
comes right after the action that changed it.
It will call.
Reading is correct. Reworded.

+ * is called.
+ *
+ * If the subscriber state has not changed, but the pause state has, the
+ * `onPauseStateChange` callback is notified as soon as possible, after firing
+ * a current event or completing a another callback. This happens if the stream

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

-a-

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

Done.

+ * is not paused, and a listener pauses it, or if the stream has been resumed
+ * from pause and has no pending events. If the listeners resume a paused stream
+ * while it still has queued events, the controller will still consider the
+ * stream paused until all queued events have been dispatched.
+ *
+ * Whether to invoke a callback depends only on the state before and after
+ * a stream action, for example firing an event. If the state changes multiple
+ * times during the action, and then ends up in the same state as before, no
+ * callback is performed.
+ *
+ * If listeners are added after the stream has completed (sent a "done" event),
+ * the listeners will be sent a "done" event eventually, but they won't affect
+ * the stream at all, and won't trigger callbacks. From the controller's point
+ * of view, the stream is completely inert when has completed.
  */
 class StreamController<T> implements StreamSink<T> {
   final _StreamImpl<T> stream;
 
   /**
    * A controller with a broadcast [stream]..
    *
    * The [onPauseStateChange] function is called when the stream becomes
    * paused or resumes after being paused. The current pause state can
    * be read from [isPaused]. Ignored if [:null:].
@@ skipping 33 matching lines @@
 
   /**
    * Whether the stream is closed for adding more events.
    *
    * If true, the "done" event might not have fired yet, but it has been
    * scheduled, and it is too late to add more events.
    */
   bool get isClosed => stream._isClosed;
 
   /** Whether one or more active subscribers have requested a pause. */
-  bool get isPaused => stream._isPaused;
+  bool get isPaused => stream._isControllerPaused;
 
   /** Whether there are currently any subscribers on this [Stream]. */
   bool get hasSubscribers => stream._hasSubscribers;
 
   /**
    * Send or queue a data event.
    */
   void add(T value) => stream._add(value);
 
   /**
@@ skipping 51 matching lines @@
   _SingleControllerStream(this._subscriptionHandler, this._pauseHandler);
 
   void _onSubscriptionStateChange() {
     if (_subscriptionHandler != null) _subscriptionHandler();
   }
 
   void _onPauseStateChange() {
     if (_pauseHandler != null) _pauseHandler();
   }
 }