Fix listeners being added to subscription after onListen was called.

sdk/lib/async/broadcast_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;
 
 class _BroadcastStream<T> extends _ControllerStream<T> {
   _BroadcastStream(_StreamControllerLifecycle controller) : super(controller);
 
   bool get isBroadcast => true;
@@ skipping 11 matching lines @@
   static const int _STATE_REMOVE_AFTER_FIRING = 4;
   // TODO(lrn): Use the _state field on _ControllerSubscription to
   // also store this state. Requires that the subscription implementation
   // does not assume that it's use of the state integer is the only use.
   int _eventState;
 
   _BroadcastSubscriptionLink _next;
   _BroadcastSubscriptionLink _previous;
 
   _BroadcastSubscription(_StreamControllerLifecycle controller,
+                         void onData(T data),
+                         Function onError,
+                         void onDone(),
                          bool cancelOnError)
-      : super(controller, cancelOnError) {
+      : super(controller, onData, onError, onDone, cancelOnError) {
     _next = _previous = this;
   }
 
   _BroadcastStreamController get _controller => super._controller;
 
   bool _expectsEvent(int eventId) =>
       (_eventState & _STATE_EVENT_ID) == eventId;
 
   void _toggleEventId() {
     _eventState ^= _STATE_EVENT_ID;
@@ skipping 127 matching lines @@
     assert(!identical(subscription._next, subscription));
     _BroadcastSubscriptionLink previous = subscription._previous;
     _BroadcastSubscriptionLink next = subscription._next;
     previous._next = next;
     next._previous = previous;
     subscription._next = subscription._previous = subscription;
   }
 
   // _StreamControllerLifecycle interface.
 
-  StreamSubscription<T> _subscribe(bool cancelOnError) {
+  StreamSubscription<T> _subscribe(
+      void onData(T data),
+      Function onError,
+      void onDone(),
+      bool cancelOnError) {
     if (isClosed) {
-      return new _DoneStreamSubscription<T>(_nullDoneHandler);
+      if (onDone == null) onDone = _nullDoneHandler;
+      return new _DoneStreamSubscription<T>(onDone);
     }
     StreamSubscription subscription =
-        new _BroadcastSubscription<T>(this, cancelOnError);
+        new _BroadcastSubscription<T>(this, onData, onError, onDone,
+                                      cancelOnError);
     _addListener(subscription);
     if (identical(_next, _previous)) {
       // Only one listener, so it must be the first listener.
       _runGuarded(_onListen);
     }
     return subscription;
   }
 
   Future _recordCancel(_BroadcastSubscription<T> subscription) {
     // If already removed by the stream, don't remove it again.
@@ skipping 291 matching lines @@
     _pauseCount++;
   }
   void resume() { _resume(null); }
   void _resume(_) {
     if (_pauseCount > 0) _pauseCount--;
   }
   Future cancel() { return new _Future.immediate(null); }
   bool get isPaused => _pauseCount > 0;
   Future asFuture([Object value]) => new _Future();
 }