Implement asBroadcast using a _MultiplexStreamController.

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 430 matching lines @@
    * A multiplex 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;
     previous._next = subscription;
     _previous = subscription._previous;
     subscription._previous._next = this;
@@ skipping 32 matching lines @@
     }
   }
 
   void _recordPause(StreamSubscription<T> subscription) {}
   void _recordResume(StreamSubscription<T> subscription) {}
 
   // EventSink interface.
 
   void add(T data) {
     assert(!isClosed);
+    _sendData(data);
+  }
+
+  void addError(Object error, [Object stackTrace]) {
+    assert(!isClosed);
+    // TODO(lrn): Handle stacktrace.

[floitsch, 2013/05/29 09:39:58]

That should be less than 5 lines of code.

[Lasse Reichstein Nielsen, 2013/05/29 10:39:12]

Done.

+    _sendError(error);
+  }
+
+  void close() {
+    assert(!isClosed);
+    _state |= _STATE_CLOSED;
+    _sendDone();
+  }
+
+  // EventDispatch interface.
+
+  void _sendData(T data) {
     if (_isEmpty) return;
     _forEachListener((_BufferingStreamSubscription<T> subscription) {
       subscription._add(data);
     });
   }
 
-  void addError(Object error, [Object stackTrace]) {
-    assert(!isClosed);
+  void _sendError(Object error) {
     if (_isEmpty) return;
     _forEachListener((_BufferingStreamSubscription<T> subscription) {
       subscription._addError(error);
     });
   }
 
-  void close() {
-    assert(!isClosed);
-    _state |= _STATE_CLOSED;
+  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 ((_state & _STATE_FIRING) != 0) {
+    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.
     //
@@ skipping 17 matching lines @@
         link = subscription._next;
       }
     }
     _state &= ~_STATE_FIRING;
 
     if (_isEmpty) {
       _runGuarded(_onCancel);
     }
   }
 }
+
+class _BufferingMultiplexStreamController<T>
+    extends _MultiplexStreamController<T>
+    implements _EventDispatch<T> {
+  _StreamImplEvents _pending;
+
+  _BufferingMultiplexStreamController(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);
+  }
+
+  void add(T data) {
+    if (_isFiring || _hasPending) {

[floitsch, 2013/05/29 09:39:58]

add comment when _hasPending but not _isFiring.

[Lasse Reichstein Nielsen, 2013/05/29 10:39:12]

It's only during the onCancel callback after removing the last listener while
firing.
I'll just make it clear all events in that case, which is consistent with not
having any listeners. That means that during onCancel, there are never pending
events.

+      _addPendingEvent(new _DelayedData<T>(data));
+      return;
+    }
+    super.add(data);
+    while (_hasPending) {
+      _pending.handleNext(this);
+    }
+  }
+
+  void addError(Object error, [StackTrace stackTrace]) {
+    if (_isFiring || _hasPending) {
+      _addPendingEvent(new _DelayedError(error));
+      return;
+    }
+    super.addError(error, stackTrace);
+    while (_hasPending) {
+      _pending.handleNext(this);
+    }
+  }
+
+  void close() {
+    if (_isFiring || _hasPending) {
+      _addPendingEvent(const _DelayedDone());
+      _state |= _STATE_CLOSED;
+      return;
+    }
+    super.close();
+    assert(!_hasPending);
+  }
+}