Fix listeners being added to subscription after onListen was called.

sdk/lib/async/stream_transformers.dart

 // Copyright (c) 2013, 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;
 
 /**
  * Wraps an [_EventSink] so it exposes only the [EventSink] interface.
  */
 class _EventSinkWrapper<T> implements EventSink<T> {
@@ skipping 17 matching lines @@
 class _SinkTransformerStreamSubscription<S, T>
     extends _BufferingStreamSubscription<T> {
   /// The transformer's input sink.
   EventSink _transformerSink;
 
   /// The subscription to the input stream.
   StreamSubscription<S> _subscription;
 
   _SinkTransformerStreamSubscription(Stream<S> source,
                                      _SinkMapper mapper,
+                                     void onData(T data),
+                                     Function onError,
+                                     void onDone(),
                                      bool cancelOnError)
       // We set the adapter's target only when the user is allowed to send data.
-      : super(cancelOnError) {
+      : super(onData, onError, onDone, cancelOnError) {
     _EventSinkWrapper<T> eventSink = new _EventSinkWrapper<T>(this);
     _transformerSink = mapper(eventSink);
     _subscription = source.listen(_handleData,
                                   onError: _handleError,
                                   onDone: _handleDone);
   }
 
   /** Whether this subscription is still subscribed to its source. */
   bool get _isSubscribed => _subscription != null;
 
@@ skipping 123 matching lines @@
   bool get isBroadcast => _stream.isBroadcast;
 
   _BoundSinkStream(this._stream, this._sinkMapper);
 
   StreamSubscription<T> listen(void onData(T event),
                                { Function onError,
                                  void onDone(),
                                  bool cancelOnError }) {
     cancelOnError = identical(true, cancelOnError);
     StreamSubscription<T> subscription = new _SinkTransformerStreamSubscription(
-        _stream, _sinkMapper, cancelOnError);
-    subscription.onData(onData);
-    subscription.onError(onError);
-    subscription.onDone(onDone);
+        _stream, _sinkMapper, onData, onError, onDone, cancelOnError);
     return subscription;
   }
 }
 
 /// Data-handler coming from [StreamTransformer.fromHandlers].
 typedef void _TransformDataHandler<S, T>(S data, EventSink<T> sink);
 /// Error-handler coming from [StreamTransformer.fromHandlers].
 typedef void _TransformErrorHandler<T>(
     Object error, StackTrace stackTrace, EventSink<T> sink);
 /// Done-handler coming from [StreamTransformer.fromHandlers].
@@ skipping 104 matching lines @@
                                  void onDone(),
                                  bool cancelOnError }) {
     cancelOnError = identical(true, cancelOnError);
     StreamSubscription<T> result = _transformer(_stream, cancelOnError);
     result.onData(onData);
     result.onError(onError);
     result.onDone(onDone);
     return result;
   }
 }