More minor type improvements in dart:async.

sdk/lib/async/stream.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;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 265 matching lines @@
    *     stringStream.transform(new DuplicationTransformer());
    *
    * The resulting stream is a broadcast stream if [source] is.
    */
   factory Stream.eventTransformed(
       Stream source, EventSink mapSink(EventSink<T> sink)) {
     return new _BoundSinkStream(source, mapSink);
   }
 
   /**
-   * Reports whether this stream is a broadcast stream.
+   * Whether this stream is a broadcast stream.
    */
   bool get isBroadcast => false;
 
   /**
    * Returns a multi-subscription stream that produces the same events as this.
    *
    * The returned stream will subscribe to this stream when its first
    * subscriber is added, and will stay subscribed until this stream ends,
    * or a callback cancels the subscription.
    *
@@ skipping 92 matching lines @@
   /**
    * Creates a new stream with each data event of this stream asynchronously
    * mapped to a new event.
    *
    * This acts like [map], except that [convert] may return a [Future],
    * and in that case, the stream waits for that future to complete before
    * continuing with its result.
    *
    * The returned stream is a broadcast stream if this stream is.
    */
-  Stream<E> asyncMap<E>(convert(T event)) {
+  Stream<E> asyncMap<E>(FutureOr<E> convert(T event)) {
     StreamController<E> controller;
     StreamSubscription<T> subscription;
 
     void onListen() {
       final add = controller.add;
       assert(controller is _StreamController ||
           controller is _BroadcastStreamController);
       final _EventSink<E> eventSink = controller as Object/*=_EventSink<E>*/;
       final addError = eventSink._addError;
       subscription = this.listen((T event) {
-        dynamic newValue;
+        FutureOr<E> newValue;
         try {
           newValue = convert(event);
         } catch (e, s) {
           controller.addError(e, s);
           return;
         }
-        if (newValue is Future) {
+        if (newValue is Future<E>) {
           subscription.pause();
           newValue
               .then(add, onError: addError)
               .whenComplete(subscription.resume);
         } else {
           controller.add(newValue as Object/*=E*/);
         }
       }, onError: addError, onDone: controller.close);
     }
 
@@ skipping 275 matching lines @@
    *
    * Completes the returned [Future] when all events of the stream
    * have been processed. Completes the future with an error if the
    * stream has an error event, or if [action] throws.
    */
   Future forEach(void action(T element)) {
     _Future future = new _Future();
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
-          _runUserCode(() => action(element), (_) {},
+          // TODO(floitsch): the type should be 'void' and inferred.
+          _runUserCode<dynamic>(() => action(element), (_) {},
               _cancelAndErrorClosure(subscription, future));
         },
         onError: future._completeError,
         onDone: () {
           future._complete(null);
         },
         cancelOnError: true);
     return future;
   }
 
@@ skipping 438 matching lines @@
             _completeWithErrorCallback(future, e, s);
           }
         },
         cancelOnError: true);
     return future;
   }
 
   /**
    * Finds the single element in this stream matching [test].
    *
-   * Like [lastMatch], except that it is an error if more than one
+   * Like [lastWhere], except that it is an error if more than one
    * matching element occurs in the stream.
    */
   Future<T> singleWhere(bool test(T element)) {
     _Future<T> future = new _Future<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
         (T value) {
           _runUserCode(() => true == test(value), (bool isMatch) {
@@ skipping 279 matching lines @@
 }
 
 /**
  * An interface that abstracts creation or handling of [Stream] events.
  */
 abstract class EventSink<T> implements Sink<T> {
   /** Send a data event to a stream. */
   void add(T event);
 
   /** Send an async error to a stream. */
-  void addError(errorEvent, [StackTrace stackTrace]);
+  void addError(Object errorEvent, [StackTrace stackTrace]);
 
   /** Close the sink. No further events can be added after closing. */
   void close();
 }
 
 /** [Stream] wrapper that only exposes the [Stream] interface. */
 class StreamView<T> extends Stream<T> {
   final Stream<T> _stream;
 
   const StreamView(Stream<T> stream)
@@ skipping 306 matching lines @@
   }
 
   void addError(error, [StackTrace stackTrace]) {
     _sink.addError(error, stackTrace);
   }
 
   void close() {
     _sink.close();
   }
 }