Make new StreamController be async by default.

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.
 // -------------------------------------------------------------------
 
@@ skipping 46 matching lines @@
    * registered.
    *
    * The [onPause] function is called when the stream becomes
    * paused. [onResume] is called when the stream resumed.
    *
    * The [onListen] callback is called when the stream
    * receives its listener and [onCancel] when the listener ends
    * its subscription.
    *
    * If the stream is canceled before the controller needs new data the
    * [onResume] call might not be executed.

[floitsch, 2013/05/30 12:13:48]

Comment `sync`.

[Lasse Reichstein Nielsen, 2013/05/31 05:51:59]

Done.

    */
   factory StreamController({void onListen(),
                             void onPause(),
                             void onResume(),
-                            void onCancel()})
-      => new _StreamControllerImpl<T>(onListen, onPause, onResume, onCancel);
+                            void onCancel(),
+                            bool sync: false})
+      => sync
+         ? new _SyncStreamController<T>(onListen, onPause, onResume, onCancel)
+         : new _AsyncStreamController<T>(onListen, onPause, onResume, onCancel);
 
   /**
    * A controller where [stream] can be listened to more than once.
    *
    * The [Stream] returned by [stream] is a broadcast stream. It can be listened
    * to more than once.
    *
    * The controller distributes any events to all currently subscribed
    * listeners.
    * It is not allowed to call [add], [addError], or [close] before a previous
    * call has returned.
    *
    * Each listener is handled independently, and if they pause, only the pausing
    * listener is affected. A paused listener will buffer events internally until
    * unpaused or canceled.
    *
    * The [onListen] callback is called when the first listener is subscribed,
    * and the [onCancel] is called when there are no longer any active listeners.
    * If a listener is added again later, after the [onCancel] was called,
    * the [onListen] will be called again.
    */
-  factory StreamController.broadcast({void onListen(), void onCancel()}) {
-    return new _MultiplexStreamController<T>(onListen, onCancel);
+  factory StreamController.broadcast({void onListen(),
+                                      void onCancel(),
+                                      bool sync: false}) {
+    return sync
+        ? new _SyncBroadcastStreamController<T>(onListen, onCancel)
+        : new _AsyncBroadcastStreamController<T>(onListen, onCancel);
   }
 
   /**
    * Returns a view of this object that only exposes the [EventSink] interface.
    */
   EventSink<T> get sink;
 
   /**
    * Whether the stream is closed for adding more events.
    *
@@ skipping 33 matching lines @@
   void _recordPause(StreamSubscription<T> subscription) {}
   void _recordResume(StreamSubscription<T> subscription) {}
   void _recordCancel(StreamSubscription<T> subscription) {}
 }
 
 /**
  * Default implementation of [StreamController].
  *
  * Controls a stream that only supports a single controller.
  */
-class _StreamControllerImpl<T> implements StreamController<T>,
-                                          _StreamControllerLifecycle<T> {
+abstract class _StreamController<T> implements StreamController<T>,
+                                               _StreamControllerLifecycle<T>,
+                                               _EventDispatch<T> {
   static const int _STATE_OPEN = 0;
   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.
   int _state = _STATE_OPEN;
 
   // Events added to the stream before it has an active subscription.
   _PendingEvents _pendingEvents = null;
 
-  _StreamControllerImpl(this._onListen,
-                        this._onPause,
-                        this._onResume,
-                        this._onCancel) {
+  _StreamController(this._onListen,
+                    this._onPause,
+                    this._onResume,
+                    this._onCancel) {
     _stream = new _ControllerStream<T>(this);
   }
 
   Stream<T> get stream => _stream;
 
   /**
    * Returns a view of this object that only exposes the [EventSink] interface.
    */
   EventSink<T> get sink => new _EventSinkView<T>(this);
 
@@ skipping 10 matching lines @@
                                    : !_isCancelled;
 
   bool get hasListener => _subscription != null;
 
   /**
    * Send or queue a data event.
    */
   void add(T value) {
     if (isClosed) throw new StateError("Adding event after close");
     if (_subscription != null) {
-      _subscription._add(value);
+      _sendData(value);
     } else if (!_isCancelled) {
       _addPendingEvent(new _DelayedData<T>(value));
     }
   }
 
   /**
    * Send or enqueue an error 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);
     }
     if (_subscription != null) {
-      _subscription._addError(error);
+      _sendError(error);
     } else if (!_isCancelled) {
       _addPendingEvent(new _DelayedError(error));
     }
   }
 
   /**
    * Closes this controller.
    *
    * After closing, no further events may be added using [add] or [addError].
    *
    * You are allowed to close the controller more than once, but only the first
    * call has any effect.
    *
    * The first time a controller is closed, a "done" event is sent to its
    * stream.
    */
   void close() {
     if (isClosed) return;
     _state |= _STATE_CLOSED;
     if (_subscription != null) {
-      _subscription._close();
+      _sendDone();
     } else if (!_isCancelled) {
       _addPendingEvent(const _DelayedDone());
     }
   }
 
+  // EventDispatch interface
+
   void _addPendingEvent(_DelayedEvent event) {
     if (_isCancelled) return;
     _StreamImplEvents events = _pendingEvents;
     if (events == null) {
       events = _pendingEvents = new _StreamImplEvents();
     }
     events.add(event);
   }
 
   void _recordListen(_BufferingStreamSubscription<T> subscription) {
@@ skipping 15 matching lines @@
 
   void _recordPause(StreamSubscription<T> subscription) {
     _runGuarded(_onPause);
   }
 
   void _recordResume(StreamSubscription<T> subscription) {
     _runGuarded(_onResume);
   }
 }
 
+class _SyncStreamController<T> extends _StreamController<T> {
+  _SyncStreamController(void onListen(),
+                        void onPause(),
+                        void onResume(),
+                        void onCancel())
+      : super(onListen, onPause, onResume, onCancel);
+
+  void _sendData(T data) {
+    _subscription._add(data);
+  }
+
+  void _sendError(Object error) {
+    _subscription._addError(error);
+  }
+
+  void _sendDone() {
+    _subscription._close();
+  }
+}
+
+class _AsyncStreamController<T> extends _StreamController<T> {
+  _AsyncStreamController(void onListen(),
+                         void onPause(),
+                         void onResume(),
+                         void onCancel())
+      : super(onListen, onPause, onResume, onCancel);
+
+  void _sendData(T data) {
+    runAsync(() {

[floitsch, 2013/05/30 12:13:48]

Try to use addPending instead.

[Lasse Reichstein Nielsen, 2013/05/31 05:51:59]

Done.

+      if (_subscription == null) return;
+      _subscription._add(data);
+    });
+  }
+
+  void _sendError(Object error) {
+    runAsync(() {
+      if (_subscription == null) return;
+      _subscription._addError(error);
+    });
+  }
+
+  void _sendDone() {
+    runAsync(_subscription._close);
+  }
+}
+
 typedef void _NotificationHandler();
 
 void _runGuarded(_NotificationHandler notificationHandler) {
   if (notificationHandler == null) return;
   try {
     notificationHandler();
   } catch (e, s) {
     _throwDelayed(e, s);
   }
 }
@@ skipping 38 matching lines @@
 
   void _onPause() {
     _controller._recordPause(this);
   }
 
   void _onResume() {
     _controller._recordResume(this);
   }
 }
 
-class _MultiplexStream<T> extends _StreamImpl<T> {
-  _MultiplexStreamController _controller;
+class _BroadcastStream<T> extends _StreamImpl<T> {
+  _BroadcastStreamController _controller;
 
-  _MultiplexStream(this._controller);
+  _BroadcastStream(this._controller);
 
   bool get isBroadcast => true;
 
   StreamSubscription<T> _createSubscription(
       void onData(T data),
       void onError(Object error),
       void onDone(),
       bool cancelOnError) {
-    return new _MultiplexSubscription<T>(
+    return new _BroadcastSubscription<T>(
         _controller, onData, onError, onDone, cancelOnError);
   }
 
   void _onListen(_BufferingStreamSubscription subscription) {
     _controller._recordListen(subscription);
   }
 }
 
-abstract class _MultiplexSubscriptionLink {
-  _MultiplexSubscriptionLink _next;
-  _MultiplexSubscriptionLink _previous;
+abstract class _BroadcastSubscriptionLink {
+  _BroadcastSubscriptionLink _next;
+  _BroadcastSubscriptionLink _previous;
 }
 
-class _MultiplexSubscription<T> extends _ControllerSubscription<T>
-                                implements _MultiplexSubscriptionLink {
+class _BroadcastSubscription<T> extends _ControllerSubscription<T>
+                                implements _BroadcastSubscriptionLink {
   static const int _STATE_EVENT_ID = 1;
   static const int _STATE_FIRING = 2;
   static const int _STATE_REMOVE_AFTER_FIRING = 4;
   int _eventState;
 
-  _MultiplexSubscriptionLink _next;
-  _MultiplexSubscriptionLink _previous;
+  _BroadcastSubscriptionLink _next;
+  _BroadcastSubscriptionLink _previous;
 
-  _MultiplexSubscription(_StreamControllerLifecycle controller,
+  _BroadcastSubscription(_StreamControllerLifecycle controller,
                          void onData(T data),
                          void onError(Object error),
                          void onDone(),
                          bool cancelOnError)
       : super(controller, onData, onError, onDone, cancelOnError) {
     _next = _previous = this;
   }
 
-  _MultiplexStreamController get _controller => super._controller;
+  _BroadcastStreamController get _controller => super._controller;
 
   bool _expectsEvent(int eventId) {
     return (_eventState & _STATE_EVENT_ID) == eventId;
   }
 
   void _toggleEventId() {
     _eventState ^= _STATE_EVENT_ID;
   }
 
   bool get _isFiring => (_eventState & _STATE_FIRING) != 0;
 
   bool _setRemoveAfterFiring() {
     assert(_isFiring);
     _eventState |= _STATE_REMOVE_AFTER_FIRING;
   }
 
   bool get _removeAfterFiring =>
       (_eventState & _STATE_REMOVE_AFTER_FIRING) != 0;
 }
 
 
-class _MultiplexStreamController<T> implements StreamController<T>,
-                                               _StreamControllerLifecycle<T>,
-                                               _MultiplexSubscriptionLink {
+abstract class _BroadcastStreamController<T>
+    implements StreamController<T>,
+               _StreamControllerLifecycle<T>,
+               _BroadcastSubscriptionLink,
+               _EventDispatch<T> {
   static const int _STATE_INITIAL = 0;
   static const int _STATE_EVENT_ID = 1;
   static const int _STATE_FIRING = 2;
   static const int _STATE_CLOSED = 4;
 
   final _NotificationHandler _onListen;
   final _NotificationHandler _onCancel;
 
   // State of the controller.
   int _state;
 
   // Double-linked list of active listeners.
-  _MultiplexSubscriptionLink _next;
-  _MultiplexSubscriptionLink _previous;
+  _BroadcastSubscriptionLink _next;
+  _BroadcastSubscriptionLink _previous;
 
-  _MultiplexStreamController(this._onListen, this._onCancel)
+  _BroadcastStreamController(this._onListen, this._onCancel)
       : _state = _STATE_INITIAL {
     _next = _previous = this;
   }
 
   // StreamController interface.
 
-  Stream<T> get stream => new _MultiplexStream<T>(this);
+  Stream<T> get stream => new _BroadcastStream<T>(this);
 
   EventSink<T> get sink => new _EventSinkView<T>(this);
 
   bool get isClosed => (_state & _STATE_CLOSED) != 0;
 
   /**
-   * A multiplex controller is never paused.
+   * A broadcast controller is never paused.
    *
    * Each receiving stream may be paused individually, and they handle their
    * own buffering.
    */
   bool get isPaused => false;
 
   /** Whether there are currently a subscriber on the [Stream]. */
   bool get hasListener => !_isEmpty;
 
   /** Whether an event is being fired (sent to some, but not all, listeners). */
   bool get _isFiring => (_state & _STATE_FIRING) != 0;
 
   // Linked list helpers
 
   bool get _isEmpty => identical(_next, this);
 
   /** Adds subscription to linked list of active listeners. */
-  void _addListener(_MultiplexSubscription<T> subscription) {
-    _MultiplexSubscriptionLink previous = _previous;
+  void _addListener(_BroadcastSubscription<T> subscription) {
+    _BroadcastSubscriptionLink previous = _previous;
     previous._next = subscription;
     _previous = subscription._previous;
     subscription._previous._next = this;
     subscription._previous = previous;
     subscription._eventState = (_state & _STATE_EVENT_ID);
   }
 
-  void _removeListener(_MultiplexSubscription<T> subscription) {
+  void _removeListener(_BroadcastSubscription<T> subscription) {
     assert(identical(subscription._controller, this));
     assert(!identical(subscription._next, subscription));
     subscription._previous._next = subscription._next;
     subscription._next._previous = subscription._previous;
     subscription._next = subscription._previous = subscription;
   }
 
   // _StreamControllerLifecycle interface.
 
-  void _recordListen(_MultiplexSubscription<T> subscription) {
+  void _recordListen(_BroadcastSubscription<T> subscription) {
     _addListener(subscription);
     if (identical(_next, _previous)) {
       // Only one listener, so it must be the first listener.
       _runGuarded(_onListen);
     }
   }
 
-  void _recordCancel(_MultiplexSubscription<T> subscription) {
+  void _recordCancel(_BroadcastSubscription<T> subscription) {
     if (subscription._isFiring) {
       subscription._setRemoveAfterFiring();
     } else {
       _removeListener(subscription);
       // If we are currently firing an event, the empty-check is performed at
       // the end of the listener loop instead of here.
       if ((_state & _STATE_FIRING) == 0 && _isEmpty) {
         _callOnCancel();
       }
     }
@@ skipping 20 matching lines @@
   }
 
   void close() {
     if (isClosed) {
       throw new StateError("Cannot add new events after calling close()");
     }
     _state |= _STATE_CLOSED;
     _sendDone();
   }
 
-  // EventDispatch interface.
-
-  void _sendData(T data) {
-    if (_isEmpty) return;
-    _forEachListener((_BufferingStreamSubscription<T> subscription) {
-      subscription._add(data);
-    });
-  }
-
-  void _sendError(Object error) {
-    if (_isEmpty) return;
-    _forEachListener((_BufferingStreamSubscription<T> subscription) {
-      subscription._addError(error);
-    });
-  }
-
-  void _sendDone() {
-    if (_isEmpty) return;
-    _forEachListener((_MultiplexSubscription<T> subscription) {
-      subscription._close();
-      subscription._eventState |=
-          _MultiplexSubscription._STATE_REMOVE_AFTER_FIRING;
-    });
-  }
-
   void _forEachListener(
       void action(_BufferingStreamSubscription<T> subscription)) {
     if (_isFiring) {
       throw new StateError(
           "Cannot fire new event. Controller is already firing an event");
     }
     if (_isEmpty) return;
 
     // Get event id of this event.
     int id = (_state & _STATE_EVENT_ID);
     // Start firing (set the _STATE_FIRING bit). We don't do [_onCancel]
     // callbacks while firing, and we prevent reentrancy of this function.
     //
     // Set [_state]'s event id to the next event's id.
     // Any listeners added while firing this event will expect the next event,
     // not this one, and won't get notified.
     _state ^= _STATE_EVENT_ID | _STATE_FIRING;
-    _MultiplexSubscriptionLink link = _next;
+    _BroadcastSubscriptionLink link = _next;
     while (!identical(link, this)) {
-      _MultiplexSubscription<T> subscription = link;
+      _BroadcastSubscription<T> subscription = link;
       if (subscription._expectsEvent(id)) {
-        subscription._eventState |= _MultiplexSubscription._STATE_FIRING;
+        subscription._eventState |= _BroadcastSubscription._STATE_FIRING;
         action(subscription);
         subscription._toggleEventId();
         link = subscription._next;
         if (subscription._removeAfterFiring) {
           _removeListener(subscription);
         }
-        subscription._eventState &= ~_MultiplexSubscription._STATE_FIRING;
+        subscription._eventState &= ~_BroadcastSubscription._STATE_FIRING;
       } else {
         link = subscription._next;
       }
     }
     _state &= ~_STATE_FIRING;
 
     if (_isEmpty) {
       _callOnCancel();
     }
   }
 
   void _callOnCancel() {
     _runGuarded(_onCancel);
   }
 }
 
-class _BufferingMultiplexStreamController<T>
-    extends _MultiplexStreamController<T>
+class _SyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {
+  _SyncBroadcastStreamController(void onListen(), void onCancel())
+      : super(onListen, onCancel);
+
+  // EventDispatch interface.
+
+  void _sendData(T data) {
+    if (_isEmpty) return;
+    _forEachListener((_BufferingStreamSubscription<T> subscription) {
+      subscription._add(data);
+    });
+  }
+
+  void _sendError(Object error) {
+    if (_isEmpty) return;
+    _forEachListener((_BufferingStreamSubscription<T> subscription) {
+      subscription._addError(error);
+    });
+  }
+
+  void _sendDone() {
+    if (_isEmpty) return;
+    _forEachListener((_BroadcastSubscription<T> subscription) {
+      subscription._close();
+      subscription._eventState |=
+          _BroadcastSubscription._STATE_REMOVE_AFTER_FIRING;
+    });
+  }
+}
+
+class _AsyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {
+  _AsyncBroadcastStreamController(void onListen(), void onCancel())
+      : super(onListen, onCancel);
+
+  // EventDispatch interface.
+
+  void _sendData(T data) {
+    runAsync(() {
+      if (_isEmpty) return;
+      _forEachListener((_BufferingStreamSubscription<T> subscription) {
+        subscription._add(data);
+      });
+    });
+  }
+
+  void _sendError(Object error) {
+    runAsync(() {
+      if (_isEmpty) return;
+      _forEachListener((_BufferingStreamSubscription<T> subscription) {
+        subscription._addError(error);
+      });
+    });
+  }
+
+  void _sendDone() {
+    runAsync(() {
+      if (_isEmpty) return;
+      _forEachListener((_BroadcastSubscription<T> subscription) {
+        subscription._close();
+        subscription._eventState |=
+            _BroadcastSubscription._STATE_REMOVE_AFTER_FIRING;
+      });
+    });
+  }
+}
+
+/**
+ * Stream controller that is used by [Stream.asBroadcastStream].
+ *
+ * This stream controller allows incoming events while it is firing
+ * other events. This is handled by delaying the events until the
+ * current event is done firing, and then fire the pending events.
+ *
+ * This class extends [_SyncBroadcastStreamController]. Events of
+ * an "asBroadcastStream" stream are always initiated by events
+ * on another stream, and it is fine to forward them synchronously.
+ */
+class _AsBroadcastStreamController<T>
+    extends _SyncBroadcastStreamController<T>
     implements _EventDispatch<T> {
   _StreamImplEvents _pending;
 
-  _BufferingMultiplexStreamController(void onListen(), void onCancel())
+  _AsBroadcastStreamController(void onListen(), void onCancel())
       : super(onListen, onCancel);
 
   bool get _hasPending => _pending != null && ! _pending.isEmpty;
 
   void _addPendingEvent(_DelayedEvent event) {
     if (_pending == null) {
       _pending = new _StreamImplEvents();
     }
     _pending.add(event);
   }
@@ skipping 29 matching lines @@
     super.close();
     assert(!_hasPending);
   }
 
   void _callOnCancel() {
     if (_hasPending) {
       _pending.clear();
       _pending = null;
     }
     super._callOnCancel();
-
   }
 }