unfork DDC's copy of most SDK libraries

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 126 matching lines @@
 
   void onData(void handleData(T event)) {
     if (handleData == null) handleData = _nullDataHandler;
     // TODO(floitsch): the return type should be 'void', and the type
     // should be inferred.
     _onData = _zone.registerUnaryCallback<dynamic, T>(handleData);
   }
 
   void onError(Function handleError) {
     if (handleError == null) handleError = _nullErrorHandler;
-    // We are not allowed to use 'void' as type argument for the generic type,
-    // so we use 'dynamic' instead.
-    _onError = _registerErrorHandler<dynamic>(handleError, _zone);
+    _onError = _registerErrorHandler(handleError, _zone);
   }
 
   void onDone(void handleDone()) {
     if (handleDone == null) handleDone = _nullDoneHandler;
     _onDone = _zone.registerCallback(handleDone);
   }
 
   void pause([Future resumeSignal]) {
     if (_isCanceled) return;
     bool wasPaused = _isPaused;
@@ skipping 841 matching lines @@
       _subscription = stream.listen(
           _onData, onError: _onError, onDone: _onDone, cancelOnError: true);
       var future = new _Future<bool>();
       _stateData = future;
       return future;
     }
     return new _Future<bool>.immediate(false);
   }
 
   Future cancel() {
-    StreamSubscription<T> subscription = _subscription;
+    StreamSubscription subscription = _subscription;
     Object stateData = _stateData;
     _stateData = null;
     if (subscription != null) {
       _subscription = null;
       if (!_isPaused) {
         _Future<bool> future = stateData as Object /*=_Future<bool>*/;
         future._asyncComplete(false);
       }
       return subscription.cancel();
     }
@@ skipping 30 matching lines @@
 class _EmptyStream<T> extends Stream<T> {
   const _EmptyStream() : super._internal();
   bool get isBroadcast => true;
   StreamSubscription<T> listen(void onData(T data),
                                {Function onError,
                                 void onDone(),
                                 bool cancelOnError}) {
     return new _DoneStreamSubscription<T>(onDone);
   }
 }