Fix listeners being added to subscription after onListen was called.

sdk/lib/async/stream_impl.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;
 
 /** Abstract and private interface for a place to put events. */
 abstract class _EventSink<T> {
   void _add(T data);
   void _addError(Object error, StackTrace stackTrace);
@@ skipping 90 matching lines @@
   /** The future [_onCancel] may return. */
   Future _cancelFuture;
 
   /**
    * Queue of pending events.
    *
    * Is created when necessary, or set in constructor for preconfigured events.
    */
   _PendingEvents _pending;
 
-  _BufferingStreamSubscription(bool cancelOnError)
-      : _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0);
+  _BufferingStreamSubscription(void onData(T data),
+                               Function onError,
+                               void onDone(),
+                               bool cancelOnError)
+      : _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) {
+    this.onData(onData);
+    this.onError(onError);
+    this.onDone(onDone);
+  }
 
   /**
    * Sets the subscription's pending events object.
    *
    * This can only be done once. The pending events object is used for the
    * rest of the subscription's life cycle.
    */
   void _setPendingEvents(_PendingEvents pendingEvents) {
     assert(_pending == null);
     if (pendingEvents == null) return;
@@ skipping 336 matching lines @@
 // -------------------------------------------------------------------
 abstract class _StreamImpl<T> extends Stream<T> {
   // ------------------------------------------------------------------
   // Stream interface.
 
   StreamSubscription<T> listen(void onData(T data),
                                { Function onError,
                                  void onDone(),
                                  bool cancelOnError }) {
     cancelOnError = identical(true, cancelOnError);
-    StreamSubscription subscription = _createSubscription(cancelOnError);
-    subscription.onData(onData);
-    subscription.onError(onError);
-    subscription.onDone(onDone);
+    StreamSubscription subscription =
+        _createSubscription(onData, onError, onDone, cancelOnError);
     _onListen(subscription);
     return subscription;
   }
 
   // -------------------------------------------------------------------
   /** Create a subscription object. Called by [subcribe]. */
-  _BufferingStreamSubscription<T> _createSubscription(bool cancelOnError) {
-    return new _BufferingStreamSubscription<T>(cancelOnError);
+  _BufferingStreamSubscription<T> _createSubscription(
+      void onData(T data),
+      Function onError,
+      void onDone(),
+      bool cancelOnError) {
+    return new _BufferingStreamSubscription<T>(onData, onError, onDone,
+                                               cancelOnError);
   }
 
   /** Hook called when the subscription has been created. */
   void _onListen(StreamSubscription subscription) {}
 }
 
 typedef _PendingEvents _EventGenerator();
 
 /** Stream that generates its own events. */
 class _GeneratedStreamImpl<T> extends _StreamImpl<T> {
   final _EventGenerator _pending;
   bool _isUsed = false;
   /**
    * Initializes the stream to have only the events provided by a
    * [_PendingEvents].
    *
    * A new [_PendingEvents] must be generated for each listen.
    */
   _GeneratedStreamImpl(this._pending);
 
-  StreamSubscription _createSubscription(bool cancelOnError) {
+  StreamSubscription _createSubscription(
+      void onData(T data),
+      Function onError,
+      void onDone(),
+      bool cancelOnError) {
     if (_isUsed) throw new StateError("Stream has already been listened to.");
     _isUsed = true;
-    _BufferingStreamSubscription<T> subscription =
-         new _BufferingStreamSubscription(cancelOnError);
-    subscription._setPendingEvents(_pending());
-    return subscription;
+    return new _BufferingStreamSubscription(
+        onData, onError, onDone, cancelOnError).._setPendingEvents(_pending());
   }
 }
 
 
 /** Pending events object that gets its events from an [Iterable]. */
 class _IterablePendingEvents<T> extends _PendingEvents {
   // The iterator providing data for data events.
   // Set to null when iteration has completed.
   Iterator<T> _iterator;
 
@@ skipping 299 matching lines @@
       // Return a dummy subscription backed by nothing, since
       // it will only ever send one done event.
       return new _DoneStreamSubscription<T>(onDone);
     }
     if (_subscription == null) {
       _subscription = _source.listen(_controller.add,
                                      onError: _controller.addError,
                                      onDone: _controller.close);
     }
     cancelOnError = identical(true, cancelOnError);
-    StreamSubscription<T> result = _controller._subscribe(cancelOnError);
-    result.onData(onData);
-    result.onError(onError);
-    result.onDone(onDone);
-    return result;
+    return _controller._subscribe(onData, onError, onDone, cancelOnError);
   }
 
   void _onCancel() {
     bool shutdown = (_controller == null) || _controller.isClosed;
     if (_onCancelHandler != null) {
       _zone.runUnary(_onCancelHandler, new _BroadcastSubscriptionWrapper(this));
     }
     if (shutdown) {
       if (_subscription != null) {
         _subscription.cancel();
@@ skipping 225 matching lines @@
       _Future<bool> hasNext = _futureOrPrefetch;
       _clear();
       hasNext._complete(false);
       return;
     }
     _subscription.pause();
     _futureOrPrefetch = null;
     _state = _STATE_EXTRA_DONE;
   }
 }