Fix listeners being added to subscription after onListen was called.

sdk/lib/async/stream_pipe.dart

Index: sdk/lib/async/stream_pipe.dart
diff --git a/sdk/lib/async/stream_pipe.dart b/sdk/lib/async/stream_pipe.dart
index 2bb6ca494207695d0421acad28a60105c08baf7e..4be169c0d9fb2b1d46376fd4e020a50be24ec20d 100644
--- a/sdk/lib/async/stream_pipe.dart
+++ b/sdk/lib/async/stream_pipe.dart
@@ -67,15 +67,16 @@ abstract class _ForwardingStream<S, T> extends Stream<T> {
                                 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.
@@ -103,8 +104,10 @@ class _ForwardingStreamSubscription<S, T>
 
   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);