Fix missing generic type parameters on streams.

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 621 matching lines @@
 
   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, onData, onError, onDone,
-                                    cancelOnError);
+        new _ControllerSubscription<T>(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 253 matching lines @@
   _StreamControllerAddStreamState(_StreamController controller,
                                   this.varData,
                                   Stream source,
                                   bool cancelOnError)
       : super(controller, source, cancelOnError) {
     if (controller.isPaused) {
       addSubscription.pause();
     }
   }
 }