Update dart:async to match the Dart repo.

tool/input_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
 // -------------------------------------------------------------------
 
+typedef void _TimerCallback();
+
 /**
  * A source of asynchronous data events.
  *
  * A Stream provides a way to receive a sequence of events.
  * Each event is either a data event or an error event,
  * representing the result of a single computation.
  * When the events provided by a Stream have all been sent,
  * a single "done" event will mark the end.
  *
  * You can [listen] on a stream to make it start generating events,
@@ skipping 20 matching lines @@
  *
  * *A broadcast stream* allows any number of listeners, and it fires
  * its events when they are ready, whether there are listeners or not.
  *
  * Broadcast streams are used for independent events/observers.
  *
  * If several listeners want to listen to a single subscription stream,
  * use [asBroadcastStream] to create a broadcast stream on top of the
  * non-broadcast stream.
  *
- * On either kind of stream, stream transformationss, such as [where] and
+ * On either kind of stream, stream transformations, such as [where] and
  * [skip], return the same type of stream as the one the method was called on,
  * unless otherwise noted.
  *
  * When an event is fired, the listener(s) at that time will receive the event.
  * If a listener is added to a broadcast stream while an event is being fired,
  * that listener will not receive the event currently being fired.
  * If a listener is canceled, it immediately stops receiving events.
  *
  * When the "done" event is fired, subscribers are unsubscribed before
  * receiving the event. After the event has been sent, the stream has no
  * subscribers. Adding new subscribers to a broadcast stream after this point
  * is allowed, but they will just receive a new "done" event as soon
  * as possible.
  *
  * Stream subscriptions always respect "pause" requests. If necessary they need
  * to buffer their input, but often, and preferably, they can simply request
  * their input to pause too.
  *
  * The default implementation of [isBroadcast] returns false.
  * A broadcast stream inheriting from [Stream] must override [isBroadcast]
  * to return `true`.
  */
 abstract class Stream<T> {
   Stream();
 
   /**
+   * Internal use only. We do not want to promise that Stream stays const.
+   *
+   * If mixins become compatible with const constructors, we may use a
+   * stream mixin instead of extending Stream from a const class.
+   */
+  const Stream._internal();
+
+  /**
+   * Creates an empty broadcast stream.
+   *
+   * This is a stream which does nothing except sending a done event
+   * when it's listened to.
+   */
+  const factory Stream.empty() = _EmptyStream<T>;
+
+  /**
    * Creates a new single-subscription stream from the future.
    *
    * When the future completes, the stream will fire one event, either
    * data or error, and then close with a done-event.
    */
   factory Stream.fromFuture(Future<T> future) {
     // Use the controller's buffering to fill in the value even before
     // the stream has a listener. For a single value, it's not worth it
     // to wait for a listener before doing the `then` on the future.
-    _StreamController<T> controller =
-        new StreamController<T>(sync: true) as _StreamController<T>;
+    _StreamController<T> controller = new StreamController<T>(sync: true);
     future.then((value) {
         controller._add(value);
         controller._closeUnchecked();
       },
       onError: (error, stackTrace) {
         controller._addError(error, stackTrace);
         controller._closeUnchecked();
       });
     return controller.stream;
   }
 
   /**
+   * Create a stream from a group of futures.
+   *
+   * The stream reports the results of the futures on the stream in the order
+   * in which the futures complete.
+   *
+   * If some futures have completed before calling `Stream.fromFutures`,
+   * their result will be output on the created stream in some unspecified
+   * order.
+   *
+   * When all futures have completed, the stream is closed.
+   *
+   * If no future is passed, the stream closes as soon as possible.
+   */
+  factory Stream.fromFutures(Iterable<Future<T>> futures) {
+    _StreamController<T> controller = new StreamController<T>(sync: true);
+    int count = 0;
+    var onValue = (T value) {
+      if (!controller.isClosed) {
+        controller._add(value);
+        if (--count == 0) controller._closeUnchecked();
+      }
+    };
+    var onError = (error, stack) {
+      if (!controller.isClosed) {
+        controller._addError(error, stack);
+        if (--count == 0) controller._closeUnchecked();
+      }
+    };
+    // The futures are already running, so start listening to them immediately
+    // (instead of waiting for the stream to be listened on).
+    // If we wait, we might not catch errors in the futures in time.
+    for (var future in futures) {
+      count++;
+      future.then(onValue, onError: onError);
+    }
+    // Use schedule microtask since controller is sync.
+    if (count == 0) scheduleMicrotask(controller.close);
+    return controller.stream;
+  }
+
+  /**
    * Creates a single-subscription stream that gets its data from [data].
    *
    * The iterable is iterated when the stream receives a listener, and stops
    * iterating if the listener cancels the subscription.
    *
    * If iterating [data] throws an error, the stream ends immediately with
    * that error. No done event will be sent (iteration is not complete), but no
    * further data events will be generated either, since iteration cannot
    * continue.
    */
   factory Stream.fromIterable(Iterable<T> data) {
     return new _GeneratedStreamImpl<T>(
         () => new _IterablePendingEvents<T>(data));
   }
 
   /**
    * Creates a stream that repeatedly emits events at [period] intervals.
    *
    * The event values are computed by invoking [computation]. The argument to
    * this callback is an integer that starts with 0 and is incremented for
    * every event.
    *
    * If [computation] is omitted the event values will all be `null`.
    */
   factory Stream.periodic(Duration period,
                           [T computation(int computationCount)]) {
-    if (computation == null) computation = ((i) => null);
-
     Timer timer;
     int computationCount = 0;
     StreamController<T> controller;
     // Counts the time that the Stream was running (and not paused).
     Stopwatch watch = new Stopwatch();
 
     void sendEvent() {
       watch.reset();
-      T data = computation(computationCount++);
+      T data;
+      if (computation != null) {
+        try {
+          data = computation(computationCount++);
+        } catch (e, s) {
+          controller.addError(e, s);
+          return;
+        }
+      }
       controller.add(data);
     }
 
     void startPeriodicTimer() {
       assert(timer == null);
       timer = new Timer.periodic(period, (Timer timer) {
         sendEvent();
       });
     }
 
@@ skipping 42 matching lines @@
    *
    *     class DuplicationSink implements EventSink<String> {
    *       final EventSink<String> _outputSink;
    *       DuplicationSink(this._outputSink);
    *
    *       void add(String data) {
    *         _outputSink.add(data);
    *         _outputSink.add(data);
    *       }
    *
-   *       void addError(e, [st]) => _outputSink(e, st);
-   *       void close() => _outputSink.close();
+   *       void addError(e, [st]) { _outputSink.addError(e, st); }
+   *       void close() { _outputSink.close(); }
    *     }
    *
    *     class DuplicationTransformer implements StreamTransformer<String, String> {
    *       // Some generic types ommitted for brevety.
-   *       Stream bind(Stream stream) => new Stream<String>.eventTransform(
+   *       Stream bind(Stream stream) => new Stream<String>.eventTransformed(
    *           stream,
    *           (EventSink sink) => new DuplicationSink(sink));
    *     }
    *
    *     stringStream.transform(new DuplicationTransformer());
    *
    * The resulting stream is a broadcast stream if [source] is.
    */
   factory Stream.eventTransformed(Stream source,
                                   EventSink mapSink(EventSink<T> sink)) {
@@ skipping 39 matching lines @@
    * is called. If [onData] is null, nothing happens.
    *
    * On errors from this stream, the [onError] handler is given a
    * object describing the error.
    *
    * The [onError] callback must be of type `void onError(error)` or
    * `void onError(error, StackTrace stackTrace)`. If [onError] accepts
    * two arguments it is called with the stack trace (which could be `null` if
    * the stream itself received an error without stack trace).
    * Otherwise it is called with just the error object.
+   * If [onError] is omitted, any errors on the stream are considered unhandled,
+   * and will be passed to the current [Zone]'s error handler.
+   * By default unhandled async errors are treated
+   * as if they were uncaught top-level errors.
    *
    * If this stream closes, the [onDone] handler is called.
    *
    * If [cancelOnError] is true, the subscription is ended when
    * the first error is reported. The default is false.
    */
   StreamSubscription<T> listen(void onData(T event),
                                { Function onError,
                                  void onDone(),
                                  bool cancelOnError});
 
   /**
    * Creates a new stream from this stream that discards some data events.
    *
    * The new stream sends the same error and done events as this stream,
    * but it only sends the data events that satisfy the [test].
    *
    * The returned stream is a broadcast stream if this stream is.
    * If a broadcast stream is listened to more than once, each subscription
    * will individually perform the `test`.
    */
   Stream<T> where(bool test(T event)) {
     return new _WhereStream<T>(this, test);
   }
 
   /**
    * Creates a new stream that converts each element of this stream
    * to a new value using the [convert] function.
    *
+   * For each data event, `o`, in this stream, the returned stream
+   * provides a data event with the value `convert(o)`.
+   * If [convert] throws, the returned stream reports the exception as an error
+   * event instead.
+   *
+   * Error and done events are passed through unchanged to the returned stream.
+   *
    * The returned stream is a broadcast stream if this stream is.
+   * The [convert] function is called once per data event per listener.
    * If a broadcast stream is listened to more than once, each subscription
-   * will individually execute `map` for each event.
+   * will individually call [convert] on each data event.
    */
   Stream/*<S>*/ map/*<S>*/(/*=S*/ convert(T event)) {
     return new _MapStream<T, dynamic/*=S*/>(this, convert);
   }
 
   /**
    * 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 asyncMap(convert(T event)) {
-    StreamController controller;
-    StreamSubscription subscription;
-    void onListen () {
+  Stream/*<E>*/ asyncMap/*<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 = controller as _EventSink<T>;
+      final _EventSink/*<E>*/ eventSink =
+          controller as Object /*=_EventSink<E>*/;
       final addError = eventSink._addError;
       subscription = this.listen(
           (T event) {
-            var newValue;
+            dynamic newValue;
             try {
               newValue = convert(event);
             } catch (e, s) {
               controller.addError(e, s);
               return;
             }
             if (newValue is Future) {
               subscription.pause();
               newValue.then(add, onError: addError)
                       .whenComplete(subscription.resume);
             } else {
-              controller.add(newValue);
+              controller.add(newValue as Object/*=E*/);
             }
           },
           onError: addError,
           onDone: controller.close
       );
     }
+
     if (this.isBroadcast) {
-      controller = new StreamController.broadcast(
+      controller = new StreamController/*<E>*/.broadcast(
         onListen: onListen,
         onCancel: () { subscription.cancel(); },
         sync: true
       );
     } else {
-      controller = new StreamController(
+      controller = new StreamController/*<E>*/(
         onListen: onListen,
         onPause: () { subscription.pause(); },
         onResume: () { subscription.resume(); },
         onCancel: () { subscription.cancel(); },
         sync: true
       );
     }
     return controller.stream;
   }
 
   /**
    * Creates a new stream with the events of a stream per original event.
    *
    * This acts like [expand], except that [convert] returns a [Stream]
    * instead of an [Iterable].
    * The events of the returned stream becomes the events of the returned
    * stream, in the order they are produced.
    *
    * If [convert] returns `null`, no value is put on the output stream,
    * just as if it returned an empty stream.
    *
    * The returned stream is a broadcast stream if this stream is.
    */
-  Stream asyncExpand(Stream convert(T event)) {
-    StreamController controller;
-    StreamSubscription subscription;
+  Stream/*<E>*/ asyncExpand/*<E>*/(Stream/*<E>*/ convert(T event)) {
+    StreamController/*<E>*/ controller;
+    StreamSubscription<T> subscription;
     void onListen() {
       assert(controller is _StreamController ||
              controller is _BroadcastStreamController);
-      final eventSink = controller as _EventSink<T>;
+      final _EventSink/*<E>*/ eventSink =
+          controller as Object /*=_EventSink<E>*/;
       subscription = this.listen(
           (T event) {
-            Stream newStream;
+            Stream/*<E>*/ newStream;
             try {
               newStream = convert(event);
             } catch (e, s) {
               controller.addError(e, s);
               return;
             }
             if (newStream != null) {
               subscription.pause();
               controller.addStream(newStream)
                         .whenComplete(subscription.resume);
             }
           },
           onError: eventSink._addError,  // Avoid Zone error replacement.
           onDone: controller.close
       );
     }
     if (this.isBroadcast) {
-      controller = new StreamController.broadcast(
+      controller = new StreamController/*<E>*/.broadcast(
         onListen: onListen,
         onCancel: () { subscription.cancel(); },
         sync: true
       );
     } else {
-      controller = new StreamController(
+      controller = new StreamController/*<E>*/(
         onListen: onListen,
         onPause: () { subscription.pause(); },
         onResume: () { subscription.resume(); },
         onCancel: () { subscription.cancel(); },
         sync: true
       );
     }
     return controller.stream;
   }
 
@@ skipping 38 matching lines @@
    *
    * The returned stream is a broadcast stream if this stream is.
    * If a broadcast stream is listened to more than once, each subscription
    * will individually call `convert` and expand the events.
    */
   Stream/*<S>*/ expand/*<S>*/(Iterable/*<S>*/ convert(T value)) {
     return new _ExpandStream<T, dynamic/*=S*/>(this, convert);
   }
 
   /**
-   * Binds this stream as the input of the provided [StreamConsumer].
+   * Pipe the events of this stream into [streamConsumer].
    *
-   * The `streamConsumer` is closed when the stream has been added to it.
+   * The events of this stream are added to `streamConsumer` using
+   * [StreamConsumer.addStream].
+   * The `streamConsumer` is closed when this stream has been successfully added
+   * to it - when the future returned by `addStream` completes without an error.
    *
    * Returns a future which completes when the stream has been consumed
    * and the consumer has been closed.
+   *
+   * The returned future completes with the same result as the future returned
+   * by [StreamConsumer.close].
+   * If the adding of the stream itself fails in some way,
+   * then the consumer is expected to be closed, and won't be closed again.
+   * In that case the returned future completes with the error from calling
+   * `addStream`.
    */
   Future pipe(StreamConsumer<T> streamConsumer) {
     return streamConsumer.addStream(this).then((_) => streamConsumer.close());
   }
 
   /**
    * Chains this stream as the input of the provided [StreamTransformer].
    *
    * Returns the result of [:streamTransformer.bind:] itself.
    *
@@ skipping 37 matching lines @@
         }
       },
       cancelOnError: true
     );
     return result;
   }
 
   /** Reduces a sequence of values by repeatedly applying [combine]. */
   Future/*<S>*/ fold/*<S>*/(var/*=S*/ initialValue,
       /*=S*/ combine(var/*=S*/ previous, T element)) {
-    _Future/*<S>*/ result = new _Future();
-    var value = initialValue;
+
+    _Future/*<S>*/ result = new _Future/*<S>*/();
+    var/*=S*/ value = initialValue;
     StreamSubscription subscription;
     subscription = this.listen(
       (T element) {
         _runUserCode(
           () => combine(value, element),
-          (newValue) { value = newValue; },
+          (/*=S*/ newValue) { value = newValue; },
           _cancelAndErrorClosure(subscription, result)
         );
       },
       onError: (e, st) {
         result._completeError(e, st);
       },
       onDone: () {
         result._complete(value);
       },
       cancelOnError: true);
@@ skipping 241 matching lines @@
    * Discards all data on the stream, but signals when it's done or an error
    * occured.
    *
    * When subscribing using [drain], cancelOnError will be true. This means
    * that the future will complete with the first error on the stream and then
    * cancel the subscription.
    *
    * In case of a `done` event the future completes with the given
    * [futureValue].
    */
-  Future drain([var futureValue]) => listen(null, cancelOnError: true)
-      .asFuture(futureValue);
+  Future/*<E>*/ drain/*<E>*/([/*=E*/ futureValue])
+      => listen(null, cancelOnError: true).asFuture/*<E>*/(futureValue);
 
   /**
    * Provides at most the first [n] values of this stream.
    *
    * Forwards the first [n] data events of this stream, and all error
    * events, to the returned stream, and ends with a done event.
    *
    * If this stream produces fewer than [count] values before it's done,
    * so will the returned stream.
    *
    * Stops listening to the stream after the first [n] elements have been
    * received.
    *
    * Internally the method cancels its subscription after these elements. This
    * means that single-subscription (non-broadcast) streams are closed and
    * cannot be reused after a call to this method.
    *
    * The returned stream is a broadcast stream if this stream is.
    * For a broadcast stream, the events are only counted from the time
    * the returned stream is listened to.
    */
   Stream<T> take(int count) {
-    return new _TakeStream(this, count);
+    return new _TakeStream<T>(this, count);
   }
 
   /**
    * Forwards data events while [test] is successful.
    *
    * The returned stream provides the same events as this stream as long
    * as [test] returns [:true:] for the event data. The stream is done
    * when either this stream is done, or when this stream first provides
    * a value that [test] doesn't accept.
    *
    * Stops listening to the stream after the accepted elements.
    *
    * Internally the method cancels its subscription after these elements. This
    * means that single-subscription (non-broadcast) streams are closed and
    * cannot be reused after a call to this method.
    *
    * The returned stream is a broadcast stream if this stream is.
    * For a broadcast stream, the events are only tested from the time
    * the returned stream is listened to.
    */
   Stream<T> takeWhile(bool test(T element)) {
-    return new _TakeWhileStream(this, test);
+    return new _TakeWhileStream<T>(this, test);
   }
 
   /**
    * Skips the first [count] data events from this stream.
    *
    * The returned stream is a broadcast stream if this stream is.
    * For a broadcast stream, the events are only counted from the time
    * the returned stream is listened to.
    */
   Stream<T> skip(int count) {
-    return new _SkipStream(this, count);
+    return new _SkipStream<T>(this, count);
   }
 
   /**
    * Skip data events from this stream while they are matched by [test].
    *
    * Error and done events are provided by the returned stream unmodified.
    *
    * Starting with the first data event where [test] returns false for the
    * event data, the returned stream will have the same events as this stream.
    *
    * The returned stream is a broadcast stream if this stream is.
    * For a broadcast stream, the events are only tested from the time
    * the returned stream is listened to.
    */
   Stream<T> skipWhile(bool test(T element)) {
-    return new _SkipWhileStream(this, test);
+    return new _SkipWhileStream<T>(this, test);
   }
 
   /**
    * Skips data events if they are equal to the previous data event.
    *
    * The returned stream provides the same events as this stream, except
-   * that it never provides two consequtive data events that are equal.
+   * that it never provides two consecutive data events that are equal.
    *
    * Equality is determined by the provided [equals] method. If that is
    * omitted, the '==' operator on the last provided data element is used.
    *
    * The returned stream is a broadcast stream if this stream is.
    * If a broadcast stream is listened to more than once, each subscription
    * will individually perform the `equals` test.
    */
   Stream<T> distinct([bool equals(T previous, T next)]) {
-    return new _DistinctStream(this, equals);
+    return new _DistinctStream<T>(this, equals);
   }
 
   /**
    * Returns the first element of the stream.
    *
    * Stops listening to the stream after the first element has been received.
    *
    * Internally the method cancels its subscription after the first element.
    * This means that single-subscription (non-broadcast) streams are closed
    * and cannot be reused after a call to this getter.
@@ skipping 32 matching lines @@
    * If an error event occurs before the first data event, the resulting future
    * is completed with that error.
    *
    * If this stream is empty (a done event occurs before the first data event),
    * the resulting future completes with a [StateError].
    */
   Future<T> get last {
     _Future<T> future = new _Future<T>();
     T result = null;
     bool foundResult = false;
-    StreamSubscription subscription;
-    subscription = this.listen(
+    listen(
       (T value) {
         foundResult = true;
         result = value;
       },
       onError: future._completeError,
       onDone: () {
         if (foundResult) {
           future._complete(result);
           return;
         }
@@ skipping 248 matching lines @@
    * This `EventSink` is only valid during the call to `onTimeout`.
    *
    * If `onTimeout` is omitted, a timeout will just put a [TimeoutException]
    * into the error channel of the returned stream.
    *
    * The returned stream is a broadcast stream if this stream is.
    * If a broadcast stream is listened to more than once, each subscription
    * will have its individually timer that starts counting on listen,
    * and the subscriptions' timers can be paused individually.
    */
-  Stream timeout(Duration timeLimit, {void onTimeout(EventSink sink)}) {
-    StreamController controller;
+  Stream<T> timeout(Duration timeLimit, {void onTimeout(EventSink<T> sink)}) {
+    StreamController<T> controller;
     // The following variables are set on listen.
     StreamSubscription<T> subscription;
     Timer timer;
     Zone zone;
-    Function timeout2;
+    _TimerCallback timeout;
 
     void onData(T event) {
       timer.cancel();
       controller.add(event);
-      timer = zone.createTimer(timeLimit, timeout2);
+      timer = zone.createTimer(timeLimit, timeout);
     }
     void onError(error, StackTrace stackTrace) {
       timer.cancel();
       assert(controller is _StreamController ||
              controller is _BroadcastStreamController);
-      var eventSink = controller as _EventSink<T>;
+      dynamic eventSink = controller;
       eventSink._addError(error, stackTrace);  // Avoid Zone error replacement.
-      timer = zone.createTimer(timeLimit, timeout2);
+      timer = zone.createTimer(timeLimit, timeout);
     }
     void onDone() {
       timer.cancel();
       controller.close();
     }
     void onListen() {
       // This is the onListen callback for of controller.
       // It runs in the same zone that the subscription was created in.
       // Use that zone for creating timers and running the onTimeout
       // callback.
       zone = Zone.current;
       if (onTimeout == null) {
-        timeout2 = () {
+        timeout = () {
           controller.addError(new TimeoutException("No stream event",
                                                    timeLimit), null);
         };
       } else {
-        onTimeout = zone.registerUnaryCallback(onTimeout);
+        // TODO(floitsch): the return type should be 'void', and the type
+        // should be inferred.
+        var registeredOnTimeout =
+            zone.registerUnaryCallback/*<dynamic, EventSink<T>>*/(onTimeout);
         _ControllerEventSinkWrapper wrapper =
             new _ControllerEventSinkWrapper(null);
-        timeout2 = () {
+        timeout = () {
           wrapper._sink = controller;  // Only valid during call.
-          zone.runUnaryGuarded(onTimeout, wrapper);
+          zone.runUnaryGuarded(registeredOnTimeout, wrapper);
           wrapper._sink = null;
         };
       }
 
       subscription = this.listen(onData, onError: onError, onDone: onDone);
-      timer = zone.createTimer(timeLimit, timeout2);
+      timer = zone.createTimer(timeLimit, timeout);
     }
     Future onCancel() {
       timer.cancel();
       Future result = subscription.cancel();
       subscription = null;
       return result;
     }
     controller = isBroadcast
-        ? new _SyncBroadcastStreamController(onListen, onCancel)
-        : new _SyncStreamController(
+        ? new _SyncBroadcastStreamController<T>(onListen, onCancel)
+        : new _SyncStreamController<T>(
               onListen,
               () {
                 // Don't null the timer, onCancel may call cancel again.
                 timer.cancel();
                 subscription.pause();
               },
               () {
                 subscription.resume();
-                timer = zone.createTimer(timeLimit, timeout2);
+                timer = zone.createTimer(timeLimit, timeout);
               },
               onCancel);
     return controller.stream;
   }
 }
 
 /**
- * A subscritption on events from a [Stream].
+ * A subscription on events from a [Stream].
  *
  * When you listen on a [Stream] using [Stream.listen],
  * a [StreamSubscription] object is returned.
  *
  * The subscription provides events to the listener,
  * and holds the callbacks used to handle the events.
  * The subscription can also be used to unsubscribe from the events,
  * or to temporarily pause the events from the stream.
  */
 abstract class StreamSubscription<T> {
   /**
-   * Cancels this subscription. It will no longer receive events.
+   * Cancels this subscription.
    *
-   * May return a future which completes when the stream is done cleaning up.
-   * This can be used if the stream needs to release some resources
-   * that are needed for a following operation,
-   * for example a file being read, that should be deleted afterwards.
-   * In that case, the file may not be able to be deleted successfully
-   * until the returned future has completed.
+   * After this call, the subscription no longer receives events.
    *
-   * The future will be completed with a `null` value.
+   * The stream may need to shut down the source of events and clean up after
+   * the subscription is canceled.
+   *
+   * Returns a future that is completed once the stream has finished
+   * its cleanup. May also return `null` if no cleanup was necessary.
+   *
+   * Typically, futures are returned when the stream needs to release resources.
+   * For example, a stream might need to close an open file (as an asynchronous
+   * operation). If the listener wants to delete the file after having
+   * canceled the subscription, it must wait for the cleanup future to complete.
+   *
+   * A returned future completes with a `null` value.
    * If the cleanup throws, which it really shouldn't, the returned future
-   * will be completed with that error.
-   *
-   * Returns `null` if there is no need to wait.
+   * completes with that error.
    */
   Future cancel();
 
   /**
    * Set or override the data event handler of this subscription.
    *
    * This method overrides the handler that has been set at the invocation of
    * [Stream.listen].
    */
   void onData(void handleData(T data));
@@ skipping 56 matching lines @@
    *
    * This method *overwrites* the existing [onDone] and [onError] callbacks
    * with new ones that complete the returned future.
    *
    * In case of an error the subscription will automatically cancel (even
    * when it was listening with `cancelOnError` set to `false`).
    *
    * In case of a `done` event the future completes with the given
    * [futureValue].
    */
-  Future asFuture([var futureValue]);
+  Future/*<E>*/ asFuture/*<E>*/([var/*=E*/ futureValue]);
 }
 
 
 /**
  * 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]);
 
   /** 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> {
-  Stream<T> _stream;
+  final Stream<T> _stream;
 
-  StreamView(this._stream);
+  const StreamView(Stream<T> stream) : _stream = stream, super._internal();
 
   bool get isBroadcast => _stream.isBroadcast;
 
-  Stream<T> asBroadcastStream({void onListen(StreamSubscription<T> subscription),
-                               void onCancel(StreamSubscription<T> subscription)})
+  Stream<T> asBroadcastStream(
+      {void onListen(StreamSubscription<T> subscription),
+       void onCancel(StreamSubscription<T> subscription)})
       => _stream.asBroadcastStream(onListen: onListen, onCancel: onCancel);
 
   StreamSubscription<T> listen(void onData(T value),
                                { Function onError,
                                  void onDone(),
                                  bool cancelOnError }) {
     return _stream.listen(onData, onError: onError, onDone: onDone,
                           cancelOnError: cancelOnError);
   }
 }
 
 
 /**
- * The target of a [Stream.pipe] call.
+ * Abstract interface for a "sink" accepting multiple entire streams.
  *
- * The [Stream.pipe] call will pass itself to this object, and then return
- * the resulting [Future]. The pipe should complete the future when it's
- * done.
+ * A consumer can accept a number of consecutive streams using [addStream],
+ * and when no further data need to be added, the [close] method tells the
+ * consumer to complete its work and shut down.
+ *
+ * This class is not just a [Sink<Stream>] because it is also combined with
+ * other [Sink] classes, like it's combined with [EventSink] in the
+ * [StreamSink] class.
+ *
+ * The [Stream.pipe] accepts a `StreamConsumer` and will pass the stream
+ * to the consumer's [addStream] method. When that completes, it will
+ * call [close] and then complete its own returned future.
  */
 abstract class StreamConsumer<S> {
   /**
    * Consumes the elements of [stream].
    *
    * Listens on [stream] and does something for each event.
    *
-   * The consumer may stop listening after an error, or it may consume
-   * all the errors and only stop at a done event.
+   * Returns a future which is completed when the stream is done being added,
+   * and the consumer is ready to accept a new stream.
+   * No further calls to [addStream] or [close] should happen before the
+   * returned future has completed.
+   *
+   * The consumer may stop listening to the stream after an error,
+   * it may consume all the errors and only stop at a done event,
+   * or it may be canceled early if the receiver don't want any further events.
+   *
+   * If the consumer stops listening because of some error preventing it
+   * from continuing, it may report this error in the returned future,
+   * otherwise it will just complete the future with `null`.
    */
   Future addStream(Stream<S> stream);
 
   /**
-   * Tell the consumer that no futher streams will be added.
+   * Tells the consumer that no further streams will be added.
    *
-   * Returns a future that is completed when the consumer is done handling
-   * events.
+   * This allows the consumer to complete any remaining work and release
+   * resources that are no longer needed
+   *
+   * Returns a future which is completed when the consumer has shut down.
+   * If cleaning up can fail, the error may be reported in the returned future,
+   * otherwise it completes with `null`.
    */
   Future close();
 }
 
 
 /**
+ * A object that accepts stream events both synchronously and asynchronously.
+ *
  * A [StreamSink] unifies the asynchronous methods from [StreamConsumer] and
  * the synchronous methods from [EventSink].
  *
  * The [EventSink] methods can't be used while the [addStream] is called.
  * As soon as the [addStream]'s [Future] completes with a value, the
  * [EventSink] methods can be used again.
  *
  * If [addStream] is called after any of the [EventSink] methods, it'll
  * be delayed until the underlying system has consumed the data added by the
  * [EventSink] methods.
  *
  * When [EventSink] methods are used, the [done] [Future] can be used to
  * catch any errors.
  *
  * When [close] is called, it will return the [done] [Future].
  */
-abstract class StreamSink<S> implements StreamConsumer<S>, EventSink<S> {
+abstract class StreamSink<S> implements EventSink<S>, StreamConsumer<S> {
   /**
-   * As [EventSink.close], but returns a future.
+   * Tells the stream sink that no further streams will be added.
+   *
+   * This allows the stream sink to complete any remaining work and release
+   * resources that are no longer needed
+   *
+   * Returns a future which is completed when the stream sink has shut down.
+   * If cleaning up can fail, the error may be reported in the returned future,
+   * otherwise it completes with `null`.
    *
    * Returns the same future as [done].
+   *
+   * The stream sink may close before the [close] method is called, either due
+   * to an error or because it is itself provding events to someone who has
+   * stopped listening. In that case, the [done] future is completed first,
+   * and the `close` method will return the `done` future when called.
+   *
+   * Unifies [StreamConsumer.close] and [EventSink.close] which both mark their
+   * object as not expecting any further events.
    */
   Future close();
 
   /**
    * Return a future which is completed when the [StreamSink] is finished.
    *
    * If the `StreamSink` fails with an error,
    * perhaps in response to adding events using [add], [addError] or [close],
    * the [done] future will complete with that error.
    *
@@ skipping 65 matching lines @@
    *             onListen: () {
    *               subscription = input.listen((data) {
    *                   // Duplicate the data.
    *                   controller.add(data);
    *                   controller.add(data);
    *                 },
    *                 onError: controller.addError,
    *                 onDone: controller.close,
    *                 cancelOnError: cancelOnError);
    *             },
-   *             onPause: subscription.pause,
-   *             onResume: subscription.resume,
-   *             onCancel: subscription.cancel,
+   *             onPause: () { subscription.pause(); },
+   *             onResume: () { subscription.resume(); },
+   *             onCancel: () { subscription.cancel(); },
    *             sync: true);
    *           return controller.stream.listen(null);
    *         });
    */
   const factory StreamTransformer(
       StreamSubscription<T> transformer(Stream<S> stream, bool cancelOnError))
-  = _StreamSubscriptionTransformer<S, T>;
+      = _StreamSubscriptionTransformer<S, T>;
 
   /**
    * Creates a [StreamTransformer] that delegates events to the given functions.
    *
    * Example use of a duplicating transformer:
    *
    *     stringStream.transform(new StreamTransformer<String, String>.fromHandlers(
    *         handleData: (String value, EventSink<String> sink) {
    *           sink.add(value);
    *           sink.add(value);  // Duplicate the incoming events.
    *         }));
    */
   factory StreamTransformer.fromHandlers({
       void handleData(S data, EventSink<T> sink),
       void handleError(Object error, StackTrace stackTrace, EventSink<T> sink),
       void handleDone(EventSink<T> sink)})
-      = _StreamHandlerTransformer<S, T>;
+          = _StreamHandlerTransformer<S, T>;
 
   /**
    * Transform the incoming [stream]'s events.
    *
    * Creates a new stream.
    * When this stream is listened to, it will start listening on [stream],
    * and generate events on the new stream based on the events from [stream].
    *
    * Subscriptions on the returned stream should propagate pause state
    * to the subscription on [stream].
    */
   Stream<T> bind(Stream<S> stream);
 }
 
 /**
- * An [Iterable] like interface for the values of a [Stream].
+ * An [Iterator] like interface for the values of a [Stream].
  *
  * This wraps a [Stream] and a subscription on the stream. It listens
  * on the stream, and completes the future returned by [moveNext] when the
  * next value becomes available.
  */
 abstract class StreamIterator<T> {
 
   /** Create a [StreamIterator] on [stream]. */
   factory StreamIterator(Stream<T> stream)
       // TODO(lrn): use redirecting factory constructor when type
       // arguments are supported.
       => new _StreamIteratorImpl<T>(stream);
 
   /**
    * Wait for the next stream value to be available.
    *
-   * It is not allowed to call this function again until the future has
-   * completed. If the returned future completes with anything except `true`,
-   * the iterator is done, and no new value will ever be available.
+   * Returns a future which will complete with either `true` or `false`.
+   * Completing with `true` means that another event has been received and
+   * can be read as [current].
+   * Completing with `false` means that the stream itearation is done and
+   * no further events will ever be available.
+   * The future may complete with an error, if the stream produces an error,
+   * which also ends iteration.
    *
-   * The future may complete with an error, if the stream produces an error.
+   * The function must not be called again until the future returned by a
+   * previous call is completed.
    */
   Future<bool> moveNext();
 
   /**
    * The current value of the stream.
    *
-   * Only valid when the future returned by [moveNext] completes with `true`
-   * as value, and only until the next call to [moveNext].
+   * Is `null` before the first call to [moveNext] and after a call to
+   * `moveNext` completes with a `false` result or an error.
+   *
+   * When a `moveNext` call completes with `true`, the `current` field holds
+   * the most recent event of the stream, and it stays like that until the next
+   * call to `moveNext`.
+   * Between a call to `moveNext` and when its returned future completes,
+   * the value is unspecified.
    */
   T get current;
 
   /**
    * Cancels the stream iterator (and the underlying stream subscription) early.
    *
    * The stream iterator is automatically canceled if the [moveNext] future
    * completes with either `false` or an error.
    *
-   * If a [moveNext] call has been made, it will complete with `false` as value,
-   * as will all further calls to [moveNext].
-   *
    * If you need to stop listening for values before the stream iterator is
    * automatically closed, you must call [cancel] to ensure that the stream
    * is properly closed.
    *
+   * If [moveNext] has been called when the iterator is cancelled,
+   * its returned future will complete with `false` as value,
+   * as will all further calls to [moveNext].
+   *
    * Returns a future if the cancel-operation is not completed synchronously.
    * Otherwise returns `null`.
    */
   Future cancel();
 }
 
 
 /**
  * Wraps an [_EventSink] so it exposes only the [EventSink] interface.
  */
 class _ControllerEventSinkWrapper<T> implements EventSink<T> {
   EventSink _sink;
   _ControllerEventSinkWrapper(this._sink);
 
   void add(T data) { _sink.add(data); }
   void addError(error, [StackTrace stackTrace]) {
     _sink.addError(error, stackTrace);
   }
   void close() { _sink.close(); }
 }