SDK fixes

tool/input_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 325 matching lines @@
     assert(!_isCanceled);
     assert(!_isPaused);
     assert(!_inCallback);
     bool wasInputPaused = _isInputPaused;
     _state |= _STATE_IN_CALLBACK;
     _zone.runUnaryGuarded(_onData, data);
     _state &= ~_STATE_IN_CALLBACK;
     _checkState(wasInputPaused);
   }
 
-  void _sendError(var error, StackTrace stackTrace) {
+  void _sendError(Object error, StackTrace stackTrace) {
     assert(!_isCanceled);
     assert(!_isPaused);
     assert(!_inCallback);
     bool wasInputPaused = _isInputPaused;
 
     void sendError() {
       // If the subscription has been canceled while waiting for the cancel
       // future to finish we must not report the error.
       if (_isCanceled && !_waitsForCancel) return;
       _state |= _STATE_IN_CALLBACK;
@@ skipping 114 matching lines @@
                                  bool cancelOnError }) {
     cancelOnError = identical(true, cancelOnError);
     StreamSubscription subscription =
         _createSubscription(onData, onError, onDone, cancelOnError);
     _onListen(subscription);
     return subscription;
   }
 
   // -------------------------------------------------------------------
   /** Create a subscription object. Called by [subcribe]. */
-  _BufferingStreamSubscription<T> _createSubscription(
+  StreamSubscription<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(
+  StreamSubscription<T> _createSubscription(
       void onData(T data),
       Function onError,
       void onDone(),
       bool cancelOnError) {
     if (_isUsed) throw new StateError("Stream has already been listened to.");
     _isUsed = true;
     return new _BufferingStreamSubscription(
         onData, onError, onDone, cancelOnError).._setPendingEvents(_pending());
   }
 }
@@ skipping 54 matching lines @@
 /** Default error handler, reports the error to the current zone's handler. */
 void _nullErrorHandler(error, [StackTrace stackTrace]) {
   Zone.current.handleUncaughtError(error, stackTrace);
 }
 
 /** Default done handler, does nothing. */
 void _nullDoneHandler() {}
 
 
 /** A delayed event on a buffering stream subscription. */
-abstract class _DelayedEvent {
+abstract class _DelayedEvent<T> {
   /** Added as a linked list on the [StreamController]. */
   _DelayedEvent next;
   /** Execute the delayed event on the [StreamController]. */
-  void perform(_EventDispatch dispatch);
+  void perform(_EventDispatch<T> dispatch);
 }
 
 /** A delayed data event. */
-class _DelayedData<T> extends _DelayedEvent {
+class _DelayedData<T> extends _DelayedEvent<T> {
   final T value;
   _DelayedData(this.value);
   void perform(_EventDispatch<T> dispatch) {
     dispatch._sendData(value);
   }
 }
 
 /** A delayed error event. */
 class _DelayedError extends _DelayedEvent {
   final error;
@@ skipping 285 matching lines @@
 class _BroadcastSubscriptionWrapper<T> implements StreamSubscription<T> {
   final _AsBroadcastStream _stream;
 
   _BroadcastSubscriptionWrapper(this._stream);
 
   void onData(void handleData(T data)) {
     throw new UnsupportedError(
         "Cannot change handlers of asBroadcastStream source subscription.");
   }
 
-  void onError(void handleError(Object data)) {
+  void onError(Function handleError) {
     throw new UnsupportedError(
         "Cannot change handlers of asBroadcastStream source subscription.");
   }
 
   void onDone(void handleDone()) {
     throw new UnsupportedError(
         "Cannot change handlers of asBroadcastStream source subscription.");
   }
 
   void pause([Future resumeSignal]) {
@@ skipping 165 matching lines @@
       _Future<bool> hasNext = _futureOrPrefetch;
       _clear();
       hasNext._complete(false);
       return;
     }
     _subscription.pause();
     _futureOrPrefetch = null;
     _state = _STATE_EXTRA_DONE;
   }
 }