Make StreamSubscription be the active part of a stream.

sdk/lib/async/stream_controller.dart

Index: sdk/lib/async/stream_controller.dart
diff --git a/sdk/lib/async/stream_controller.dart b/sdk/lib/async/stream_controller.dart
index 3e7dbdc2c56180f14c5920a5f94f990dd405be7d..a05af91b4ab5f70499f735ea9dc74a8ebadd033d 100644
--- a/sdk/lib/async/stream_controller.dart
+++ b/sdk/lib/async/stream_controller.dart
@@ -47,14 +47,27 @@ part of dart.async;
  * of view, the stream is completely inert when has completed.
  */
 class StreamController<T> extends EventSink<T> {
-  final _StreamImpl<T> stream;
+  static const int _STATE_CANCELLED = 1;
+  static const int _STATE_CLOSED = 2;
+
+  final _NotificationHandler _onListen;
+  final _NotificationHandler _onPause;
+  final _NotificationHandler _onResume;
+  final _NotificationHandler _onCancel;
+  _StreamImpl<T> _stream;
+
+  // An active subscription on the stream, or null if no subscripton is active.
+  _ControllerSubscription<T> _subscription;
+
+  // Whether we have sent a "done" event.

[floitsch, 2013/05/22 16:26:29]

"Wether" is not right.
Also, what does "0" mean?

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

I see "Whether"?
Will name a constant _STATE_OPEN or _STATE_INITIAL.

[floitsch, 2013/05/24 13:53:41]

I meant that this doesn't look a boolean. "Whether" is used for booleans.

+  int _state = 0;
+
+  // Events added to the stream before it has an active subscription.
+  _PendingEvents _pendingEvents = null;
 
   /**
-   *
    * If the stream is canceled before the controller needs new data the
    * [onResume] call might not be executed.
-      : stream = new _MultiControllerStream<T>(
-            onListen, onPause, onResume, onCancel);

[floitsch, 2013/05/22 16:26:29]

You will have to merge this. I removed lines 53 to 57.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

ACK.

    * A controller with a [stream] that supports only one single subscriber.
    *
    * The controller will buffer all incoming events until the subscriber is
@@ -74,8 +87,14 @@ class StreamController<T> extends EventSink<T> {
                     void onPause(),
                     void onResume(),
                     void onCancel()})
-      : stream = new _SingleControllerStream<T>(
-            onListen, onPause, onResume, onCancel);
+      : _onListen = onListen,
+        _onPause = onPause,
+        _onResume = onResume,
+        _onCancel = onCancel {
+    _stream = new _ControllerStream<T>(this);
+  }
+
+  Stream<T> get stream => _stream;
 
   /**
    * Returns a view of this object that only exposes the [EventSink] interface.
@@ -83,37 +102,56 @@ class StreamController<T> extends EventSink<T> {
   EventSink<T> get sink => new _EventSinkView<T>(this);
 
   /**
+   * Whether a listener has existed and been cancelled.
+   *
+   * After this, adding more events will be ignored.
+   */
+  bool get _isCancelled => (_state & _STATE_CANCELLED) != 0;
+
+  /**
    * 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;
+  bool get isClosed => (_state & _STATE_CLOSED) != 0;
 
-  /** Whether one or more active subscribers have requested a pause. */
-  bool get isPaused => stream._isInputPaused;
+  /** Whether the subscription is active and paused. */
+  bool get isPaused => _subscription != null && _subscription._isInputPaused;
 
-  /** Whether there are currently any subscribers on this [Stream]. */
-  bool get hasListener => stream._hasListener;
+  /** Whether there are currently any subscribers on the [Stream]. */

[floitsch, 2013/05/22 16:26:29]

there is currently a subscriber on the [Stream].

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Done.

+  bool get hasListener => _subscription != null;
 
   /**
    * Send or queue a data event.
    */
-  void add(T value) => stream._add(value);
+  void add(T value) {
+    if (isClosed) throw new StateError("Adding event after close");
+    if (_subscription != null) {
+      _subscription._add(value);
+    } else if (!_isCancelled) {
+      _addPendingEvent(new _DelayedData<T>(value));
+    }
+  }
 
   /**
    * Send or enqueue an error event.
    *
-   * If a subscription has requested to be unsubscribed on errors,
+   * If the subscription has requested to be unsubscribed on errors,

[floitsch, 2013/05/22 16:26:29]

Should we keep this here? It's not really a Controller thing.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

I'm fine with removing it.

    * it will be unsubscribed after receiving this event.
    */
   void addError(Object error, [Object stackTrace]) {
+    if (isClosed) throw new StateError("Adding event after close");
     if (stackTrace != null) {
       // Force stack trace overwrite. Even if the error already contained
       // a stack trace.
       _attachStackTrace(error, stackTrace);
     }
-    stream._addError(error);
+    if (_subscription != null) {
+      _subscription._addError(error);
+    } else if (!_isCancelled) {
+      _addPendingEvent(new _DelayedError(error));
+    }
   }
 
   /**
@@ -122,37 +160,111 @@ class StreamController<T> extends EventSink<T> {
    * The "done" message should be sent at most once by a stream, and it

[floitsch, 2013/05/22 16:26:29]

Too many "should"s.
Maybe:
Closes this controller.

It is valid to close a controller multiple times. Only the first invocation has
an effect.

Upon the first closing the controller will send (or enqueue) a done event.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

sounds fine.

    * should be the last message sent.
    */
-  void close() { stream._close(); }
+  void close() {
+    if (isClosed) return;
+    _state |= _STATE_CLOSED;
+    if (_subscription != null) {
+      _subscription._close();
+    } else if (!_isCancelled) {
+      _addPendingEvent(const _DelayedDone());
+    }
+  }
+
+  void _addPendingEvent(_DelayedEvent event) {
+    if (_isCancelled) return;
+    _StreamImplEvents events = _pendingEvents;
+    if (events == null) {
+      events = _pendingEvents = new _StreamImplEvents();
+    }
+    events.add(event);
+  }
+
+  void _recordListen(_BufferingStreamSubscription subscription) {
+    assert(_subscription == null);
+    _subscription = subscription;
+    _pendingEvents = null;  // These have been taken over by the stream.

[floitsch, 2013/05/22 16:26:29]

by the subscription.

I would prefer if we transferred the pendingEvents explicitly here. It's almost
impossible to see at the construction site.
Something like: 
subscription._transferPendingEvents(_pendingEvents);
_pendingEvents = null;

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

ok, will change to not pass them in the constructor.

+    _subscription._guardCallback(() {
+      _runGuarded(_onListen);
+    });
+  }
+
+  void _recordCancel() {
+    _subscription = null;
+    _state |= _STATE_CANCELLED;
+    _runGuarded(_onCancel);
+  }
+
+  void _recordPause() {
+    _runGuarded(_onPause);
+  }
+
+  void _recordResume() {
+    _runGuarded(_onResume);
+  }
 }
 
 typedef void _NotificationHandler();
 
-class _SingleControllerStream<T> extends _SingleStreamImpl<T> {
-  _NotificationHandler _onListen;
-  _NotificationHandler _onPause;
-  _NotificationHandler _onResume;
-  _NotificationHandler _onCancel;
-
-  // TODO(floitsch): share this code with _MultiControllerStream.
-  _runGuarded(_NotificationHandler notificationHandler) {
-    if (notificationHandler == null) return;
-    try {
-      notificationHandler();
-    } catch (e, s) {
-      _throwDelayed(e, s);
+void _runGuarded(_NotificationHandler notificationHandler) {
+  if (notificationHandler == null) return;
+  try {
+    notificationHandler();
+  } catch (e, s) {
+    _throwDelayed(e, s);
+  }
+}
+
+class _ControllerStream<T> extends _StreamImpl<T> {
+  StreamController _controller;
+  bool _hasListener = false;
+
+  _ControllerStream(this._controller);
+
+  StreamSubscription<T> _createSubscription(
+      void onData(T data),
+      void onError(Object error),
+      void onDone(),
+      bool cancelOnError) {
+    if (_hasListener) {
+      try {
+        throw 0;
+      } catch (e, s) {
+        print("LISTEN TWICE(#$hashCode)\n$s");
+      }
+      throw new StateError("The stream has already been listened to.");
     }
+    //try { throw 0; } catch (e, s) { print("LISTEN ONCE(#$hashCode):\n$s"); }
+    _hasListener = true;
+    return new _ControllerSubscription<T>(
+        _controller, onData, onError, onDone, cancelOnError);
   }
 
-  _SingleControllerStream(this._onListen,
-                          this._onPause,
-                          this._onResume,
-                          this._onCancel);
+  void _onListen(_BufferingStreamSubscription subscription) {
+    _controller._recordListen(subscription);
+  }
+}
+
+class _ControllerSubscription<T> extends _BufferingStreamSubscription<T> {
+  final StreamController _controller;
+
+  _ControllerSubscription(StreamController controller,
+                          void onData(T data),
+                          void onError(Object error),
+                          void onDone(),
+                          bool cancelOnError)
+      : _controller = controller,
+        super(onData, onError, onDone, cancelOnError,
+              controller._pendingEvents);
+
+  void _onCancel() {
+    _controller._recordCancel();
+  }
 
-  void _onSubscriptionStateChange() {
-    _runGuarded(_hasListener ? _onListen : _onCancel);
+  void _onPause() {
+    _controller._recordPause();
   }
 
-  void _onPauseStateChange() {
-    _runGuarded(_isPaused ? _onPause : _onResume);
+  void _onResume() {
+    _controller._recordResume();
   }
 }