Fix listeners being added to subscription after onListen was called.

sdk/lib/async/stream_pipe.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;
 
 /** Runs user code and takes actions depending on success or failure. */
 _runUserCode(userCode(),
              onSuccess(value),
              onError(error, StackTrace stackTrace)) {
@@ skipping 49 matching lines @@
 
   _ForwardingStream(this._source);
 
   bool get isBroadcast => _source.isBroadcast;
 
   StreamSubscription<T> listen(void onData(T value),
                               { Function onError,
                                 void onDone(),
                                 bool cancelOnError }) {
     cancelOnError = identical(true, cancelOnError);
-    StreamSubscription<T> result = _createSubscription(cancelOnError);
-    result.onData(onData);
-    result.onError(onError);
-    result.onDone(onDone);
-    return result;
+    return _createSubscription(onData, onError, onDone, cancelOnError);
   }
 
-  StreamSubscription<T> _createSubscription(bool cancelOnError) {
-    return new _ForwardingStreamSubscription<S, T>(this, cancelOnError);
+  StreamSubscription<T> _createSubscription(
+      void onData(T data),
+      Function onError,
+      void onDone(),
+      bool cancelOnError) {
+    return new _ForwardingStreamSubscription<S, T>(
+        this, onData, onError, onDone, cancelOnError);
   }
 
   // Override the following methods in subclasses to change the behavior.
 
   void _handleData(S data, _EventSink<T> sink) {
     var outputData = data;
     sink._add(outputData);
   }
 
   void _handleError(error, StackTrace stackTrace, _EventSink<T> sink) {
     sink._addError(error, stackTrace);
   }
 
   void _handleDone(_EventSink<T> sink) {
     sink._close();
   }
 }
 
 /**
  * Abstract superclass for subscriptions that forward to other subscriptions.
  */
 class _ForwardingStreamSubscription<S, T>
     extends _BufferingStreamSubscription<T> {
   final _ForwardingStream<S, T> _stream;
 
   StreamSubscription<S> _subscription;
 
-  _ForwardingStreamSubscription(this._stream, bool cancelOnError)
-      : super(cancelOnError) {
+  _ForwardingStreamSubscription(this._stream, void onData(T data),
+                                Function onError, void onDone(),
+                                bool cancelOnError)
+      : super(onData, onError, onDone, cancelOnError) {
     _subscription = _stream._source.listen(_handleData,
                                            onError: _handleError,
                                            onDone: _handleDone);
   }
 
   // _StreamSink interface.
   // Transformers sending more than one event have no way to know if the stream
   // is canceled or closed after the first, so we just ignore remaining events.
 
   void _add(T data) {
@@ skipping 280 matching lines @@
         sink._addError(e, s);
         return null;
       }
       if (!isEqual) {
         sink._add(inputEvent);
         _previous = inputEvent;
       }
     }
   }
 }