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 20 matching lines @@
   }
 
   void _recordCancel(_MultiplexSubscription<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) {
-        _runGuarded(_onCancel);
+        _callOnCancel();
       }
     }
   }
 
   void _recordPause(StreamSubscription<T> subscription) {}
   void _recordResume(StreamSubscription<T> subscription) {}
 
   // EventSink interface.
 
   void add(T data) {
-    assert(!isClosed);
+    if (isClosed) {
+      throw new StateError("Cannot add new events after calling close()");
+    }
+    _sendData(data);
+  }
+
+  void addError(Object error, [Object stackTrace]) {
+    if (isClosed) {
+      throw new StateError("Cannot add new events after calling close()");
+    }
+    if (stackTrace != null) _attachStackTrace(error, stackTrace);
+    _sendError(error);
+  }
+
+  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 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 13 matching lines @@
           _removeListener(subscription);
         }
         subscription._eventState &= ~_MultiplexSubscription._STATE_FIRING;
       } else {
         link = subscription._next;
       }
     }
     _state &= ~_STATE_FIRING;
 
     if (_isEmpty) {
-      _runGuarded(_onCancel);
+      _callOnCancel();
     }
   }
+
+  void _callOnCancel() {
+    _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) {
+      _addPendingEvent(new _DelayedData<T>(data));
+      return;
+    }
+    super.add(data);
+    while (_hasPending) {
+      _pending.handleNext(this);
+    }
+  }
+
+  void addError(Object error, [StackTrace stackTrace]) {
+    if (_isFiring) {
+      _addPendingEvent(new _DelayedError(error));
+      return;
+    }
+    super.addError(error, stackTrace);
+    while (_hasPending) {
+      _pending.handleNext(this);
+    }
+  }
+
+  void close() {
+    if (_isFiring) {
+      _addPendingEvent(const _DelayedDone());
+      _state |= _STATE_CLOSED;
+      return;
+    }
+    super.close();
+    assert(!_hasPending);
+  }
+
+  void _callOnCancel() {
+    if (_hasPending) {
+      _pending.clear();
+      _pending = null;
+    }
+    super._callOnCancel();
+
+  }
+}