Fix listeners being added to subscription after onListen was called.

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 178 matching lines @@
    * If [cancelOnError] is true, only the first error on [source] is
    * forwarded to the controller's stream, and the `addStream` ends
    * after this. If [cancelOnError] is false, all errors are forwarded
    * and only a done event will end the `addStream`.
    */
   Future addStream(Stream<T> source, {bool cancelOnError: true});
 }
 
 
 abstract class _StreamControllerLifecycle<T> {
-  StreamSubscription<T> _subscribe(bool cancelOnError);
+  StreamSubscription<T> _subscribe(
+      void onData(T data),
+      Function onError,
+      void onDone(),
+      bool cancelOnError);
   void _recordPause(StreamSubscription<T> subscription) {}
   void _recordResume(StreamSubscription<T> subscription) {}
   Future _recordCancel(StreamSubscription<T> subscription) => null;
 }
 
 /**
  * Default implementation of [StreamController].
  *
  * Controls a stream that only supports a single controller.
  */
@@ skipping 254 matching lines @@
     // End of addStream stream.
     assert(_isAddingStream);
     _StreamControllerAddStreamState addState = _varData;
     _varData = addState.varData;
     _state &= ~_STATE_ADDSTREAM;
     addState.complete();
   }
 
   // _StreamControllerLifeCycle interface
 
-  StreamSubscription<T> _subscribe(bool cancelOnError) {
+  StreamSubscription<T> _subscribe(
+      void onData(T data),
+      Function onError,
+      void onDone(),
+      bool cancelOnError) {
     if (!_isInitialState) {
       throw new StateError("Stream has already been listened to.");
     }
     _ControllerSubscription subscription =
-        new _ControllerSubscription(this, cancelOnError);
+        new _ControllerSubscription(this, onData, onError, onDone,
+                                    cancelOnError);
 
     _PendingEvents pendingEvents = _pendingEvents;
     _state |= _STATE_SUBSCRIBED;
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       addState.varData = subscription;
       addState.resume();
     } else {
       _varData = subscription;
     }
@@ skipping 156 matching lines @@
   } catch (e, s) {
     Zone.current.handleUncaughtError(e, s);
   }
 }
 
 class _ControllerStream<T> extends _StreamImpl<T> {
   _StreamControllerLifecycle<T> _controller;
 
   _ControllerStream(this._controller);
 
-  StreamSubscription<T> _createSubscription(bool cancelOnError) =>
-    _controller._subscribe(cancelOnError);
+  StreamSubscription<T> _createSubscription(
+      void onData(T data),
+      Function onError,
+      void onDone(),
+      bool cancelOnError) =>
+    _controller._subscribe(onData, onError, onDone, cancelOnError);
 
   // Override == and hashCode so that new streams returned by the same
   // controller are considered equal. The controller returns a new stream
   // each time it's queried, but doesn't have to cache the result.
 
   int get hashCode => _controller.hashCode ^ 0x35323532;
 
   bool operator==(Object other) {
     if (identical(this, other)) return true;
     if (other is! _ControllerStream) return false;
     _ControllerStream otherStream = other;
     return identical(otherStream._controller, this._controller);
   }
 }
 
 class _ControllerSubscription<T> extends _BufferingStreamSubscription<T> {
   final _StreamControllerLifecycle<T> _controller;
 
-  _ControllerSubscription(this._controller, bool cancelOnError)
-      : super(cancelOnError);
+  _ControllerSubscription(this._controller, void onData(T data),
+                          Function onError, void onDone(), bool cancelOnError)
+      : super(onData, onError, onDone, cancelOnError);
 
   Future _onCancel() {
     return _controller._recordCancel(this);
   }
 
   void _onPause() {
     _controller._recordPause(this);
   }
 
   void _onResume() {
@@ skipping 80 matching lines @@
   _StreamControllerAddStreamState(_StreamController controller,
                                   this.varData,
                                   Stream source,
                                   bool cancelOnError)
       : super(controller, source, cancelOnError) {
     if (controller.isPaused) {
       addSubscription.pause();
     }
   }
 }