Update stream documentation.

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.
+ * Each event is either a data event, also called an *element* of the stream,

[floitsch, 2017/05/24 13:37:51]

isn't `_element_` the better way for this?

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

It's equivalent in commonmark.
http://spec.commonmark.org/0.27/#emphasis-and-strong-emphasis

There is a distinction between `_` and `*`, but only in some tricky cases. In
the plain case of emphasizing a single word, they do exactly the same.
(And I tend to read `*` as emphasis and `_` as underline, so I prefer `*`).

+ * or an error event, which is a notification that something has failed.
+ * When a stream has emitted all its event,
+ * a single "done" event will notify the listener that the end has been reached.
  *
- * You can [listen] on a stream to make it start generating events,
+ * You [listen] on a stream to make it start generating events,
  * and to set up listeners that receive the events.
  * When you listen, you receive a [StreamSubscription] object
  * which is the active object providing the events,
  * and which can be used to stop listening again,
  * or to temporarily pause events from the subscription.
  *
  * There are two kinds of streams: "Single-subscription" streams and
  * "broadcast" streams.
  *
  * *A single-subscription stream* allows only a single listener during the whole
@@ skipping 78 matching lines @@
       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.
+   * Each future provides either a data event or an error event,
+   * depending on how the future completes.
    *
-   * If some futures have completed before calling `Stream.fromFutures`,
-   * their result will be output on the created stream in some unspecified
-   * order.
+   * If some futures have already completed when `Stream.fromFutures` is called,
+   * their results will be emitted 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.
+   * If [futures] is empty, 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;
+    // Declare as variable holding closure instead of as a function declaration.

[floitsch, 2017/05/24 13:37:51]

holding a closure

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

And more rewrite of that broken sentence.

+    // This avoids creating a new closure for each future.
     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();
@@ skipping 192 matching lines @@
    * 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 and sends a done event, the [onDone] handler is
    * called. If [onDone] is `null`, nothing happens.
    *
-   * If [cancelOnError] is true, the subscription is automatically cancelled
+   * If [cancelOnError] is true, the subscription is automatically canceled
    * when the first error event is delivered. The default is `false`.
    *
-   * While a subscription is paused, or when it has been cancelled,
+   * While a subscription is paused, or when it has been canceled,
    * the subscription doesn't receive events and none of the
    * event handler functions are called.
    */
   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.
+   * Creates a new stream from this stream that discards some elements.
    *
    * The new stream sends the same error and done events as this stream,
    * but it only sends the data events that satisfy the [test].
    *
+   * If the [test] function throws, the data event is dropped and the
+   * error emitted on the returned stream instead.

[floitsch, 2017/05/24 13:37:51]

error is emitted

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

I think it can read without the "is", but it's probably simple with.

+   *
    * 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);
   }
 
   /**
+   * Transform each element of this stream into a new stream event.

[floitsch, 2017/05/24 13:37:50]

Transforms

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

Done.

+   *
    * Creates a new stream that converts each element of this stream
-   * to a new value using the [convert] function.
+   * to a new value using the [convert] function, and emits the result.
    *
    * 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
+   * If [convert] throws, the returned stream reports it 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 call [convert] on each data event.
    */
   Stream<S> map<S>(S convert(T event)) {
@@ skipping 125 matching lines @@
           onResume: () {
             subscription.resume();
           },
           onCancel: () => subscription.cancel(),
           sync: true);
     }
     return controller.stream;
   }
 
   /**
-   * Creates a new stream with the events of a stream per original event.
+   * Transform each element into a sequence of asynchronous events.

[floitsch, 2017/05/24 13:37:51]

Transforms

+   *
+   * Returns a new stream that emits the error events and done event

[floitsch, 2017/05/24 13:37:51]

I think we should start with what happens to the elements. We need to say what
happens to errors and done, but it should be in a separate later sentence.

For example:
---
Returns a new stream where each element of this stream is used to create a new
stream (via [convert]). The new stream's events (elements and errors) are added
individually to the returned stream. Error events and the done event of this
stream are passed through, verbatim.

For every element, this stream pauses its input until the new stream is done.
---

Alternatively, we could adopt a more formal approach.
For example:

---
Returns a stream which is the result of transforming all events:
- each element `e` is passed to [convert]. If this operation throws, the
resulting exception is passed as error event. Otherwise, this stream is paused
and all events (except done) are passed verbatim. Once the new stream is done,
this stream is resumed.
- errors are passed through.
- done is passed through.
---

I like the idea of having a nice structure, but I'm not really happy with my
proposal yet.

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

I've tried doing a rewording, but I'm not completely satisfied either.

It gets complicated by having both "the returned stream" and "the stream
returned by [convert]". Too many returned streams.

+   * of this stream, and for each element of this stream, the returned
+   * stream emits a sequence of events, provided as a stream by the
+   * [convert] function called on the element.
    *
    * 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.
+   * The events of the stream returned by [convert] are emitted on
+   * the stream returned by [asyncExpand] 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.
+   * Further events from this stream are delayed until the stream
+   * returned by [convert] has ended.
+   *
+   * If [convert] returns `null`, no value is put on the output stream
+   * for that element, just as if it had returned an empty stream.
    *
    * The returned stream is a broadcast stream if this stream is.
    */
   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<E> eventSink = controller as Object/*=_EventSink<E>*/;
@@ skipping 48 matching lines @@
    * trace. The stack trace argument might be `null` if the stream itself
    * received an error without stack trace.
    *
    * An asynchronous error `error` is matched by a test function if
    *`test(error)` returns true. If [test] is omitted, every error is considered
    * matching.
    *
    * If the error is intercepted, the [onError] function can decide what to do
    * with it. It can throw if it wants to raise a new (or the same) error,
    * or simply return to make the stream forget the error.
+   * If the received `error` value is thrown again by the [onError] function,
+   * it acts like a `rethrow` and it is emitted along with its original
+   * stack trace, not the stack trace of the `throw` inside [onError].
    *
    * If you need to transform an error into a data event, use the more generic
    * [Stream.transform] to handle the event by writing a data event to
    * the output sink.
    *
    * 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` and handle the error.
    */
   Stream<T> handleError(Function onError, {bool test(error)}) {
     return new _HandleErrorStream<T>(this, onError, test);
   }
 
   /**
-   * Creates a new stream from this stream that converts each element
-   * into zero or more events.
+   * Transform each element of this stream into a sqeuence of elements.

[floitsch, 2017/05/24 13:37:50]

Transforms ... sequence ... 

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

Done.

    *
-   * Each incoming event is converted to an [Iterable] of new events,
-   * and each of these new events are then sent by the returned stream
-   * in order.
+   * Returns a new stream which emits the same error and done events as

[floitsch, 2017/05/24 13:37:51]

See above.

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

Reworded. 
It's simpler than the async version.

+   * this stream, and for each element of this stream,
+   * it calls [convert] with the element to get an iterable.
+   * Then each element of the iterable is emitted as an element
+   * of the returned stream, in iteration order.
+   *
+   * If iteration throws, that error is emitted as an error on the returned
+   * stream and iteration ends for that element of this 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 individually call `convert` and expand the events.
    */
-  Stream<S> expand<S>(Iterable<S> convert(T value)) {
+  Stream<S> expand<S>(Iterable<S> convert(T element)) {
     return new _ExpandStream<T, S>(this, convert);
   }
 
   /**
    * Pipe the events of this stream into [streamConsumer].

[floitsch, 2017/05/24 13:37:50]

Pipes

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

Done.

    *
-   * The events of this stream are added to `streamConsumer` using
+   * All 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`.
+   * If the call to [StreamConsumer.addStream] fails in some way, this

[floitsch, 2017/05/24 13:37:50]

"fails" -> "throws"

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

Actually, I meant the completely vague thing here, because addStream is not a
function I'm in control of.
If it decides to surface an uncaught error, so will this call.
If it puts an error into the future it returns, so will this call.

Throwing is only one failure mode for an async operation.

+   * method fails in the same way.
    */
   Future pipe(StreamConsumer<T> streamConsumer) {
     return streamConsumer.addStream(this).then((_) => streamConsumer.close());
   }
 
   /**
-   * Chains this stream as the input of the provided [StreamTransformer].
+   * Applies a [StreamTransformer] to the current stream.
    *
-   * Returns the result of [:streamTransformer.bind:] itself.
+   * Returns the result of the stream transformation,
+   * that is, the result of `streamTransformer.bind(this)`.
+   * This method simply allows writing the call to `streamTransformer.bind`
+   * in a chained fashion, like
+   * ```
+   * stream.map(mapping).transform(transformation).toList()
+   * ```
+   * which can be more convenient than calling `bind` directly.
    *
-   * The `streamTransformer` can decide whether it wants to return a
-   * broadcast stream or not.
+   * The [streamTransformer] can return any stream,

[floitsch, 2017/05/24 13:37:50]

any stream.
Whether the ...

Maybe mention, that transform can be used as a building block for all
transforming methods, and that it is usually preferable to use existing methods
if they perform the intended task.

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

Done.

+   * so whether the returned stream is a broadcast stream or not,
+   * and which elements it will contain,
+   * is entirely up to the transformation.
    */
   Stream<S> transform<S>(StreamTransformer<T, S> streamTransformer) {
     return streamTransformer.bind(this);
   }
 
   /**
-   * Reduces a sequence of values by repeatedly applying [combine].
+   * Combines a sequence of values by repeatedly applying [combine].
+   *
+   * Similar to [Iterable.reduce], this function maintains a value,
+   * starting with the first element of the stream
+   * and updated for each further element of this stream.
+   * For each element after the first,
+   * the value is updated to the result of calling [combine]
+   * with the previous value and the element.
+   *
+   * When this stream is done, the returned future is completed with
+   * the value at that time.
+   *
+   * If the stream is empty, the returned future is completed with
+   * an error.
+   * If this stream emits an error, or the call to [combine] throws,
+   * the returned future is completed with that error,
+   * and processing is stopped.
    */
   Future<T> reduce(T combine(T previous, T element)) {
     _Future<T> result = new _Future<T>();
     bool seenFirst = false;
     T value;
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
           if (seenFirst) {
             _runUserCode(() => combine(value, element), (T newValue) {
               value = newValue;
             }, _cancelAndErrorClosure(subscription, result));
           } else {
             value = element;
             seenFirst = true;
           }
         },
         onError: result._completeError,
         onDone: () {
           if (!seenFirst) {
             try {
+              // Throw and recatch, instead of just doing
+              //  _completeWithErrorCallback, e, theError, StackTrace.current),
+              // to ensure that the stackTrace is set on the error.
               throw IterableElementError.noElement();
             } catch (e, s) {
               _completeWithErrorCallback(result, e, s);
             }
           } else {
             result._complete(value);
           }
         },
         cancelOnError: true);
     return result;
   }
 
-  /** Reduces a sequence of values by repeatedly applying [combine]. */
+  /**
+   * Combines a sequence of values by repeatedly applying [combine].
+   *
+   * Similar to [Iterable.fold], this function maintains a value,
+   * starting with [initialValue] and updated for each element of
+   * this stream.
+   * For each element, the value is updated to the result of calling
+   * [combine] with the previous value and the element.
+   *
+   * When this stream is done, the returned future is completed with
+   * the value at that time.
+   * For an empty stream, the future is completed with [initialValue].
+   *
+   * If this stream emits an error, or the call to [combine] throws,
+   * the returned future is completed with that error,
+   * and processing is stopped.
+   */
   Future<S> fold<S>(S initialValue, S combine(S previous, T element)) {
     _Future<S> result = new _Future<S>();
     S value = initialValue;
     StreamSubscription subscription;
-    subscription = this.listen((T element) {
-      _runUserCode(() => combine(value, element), (S newValue) {
-        value = newValue;
-      }, _cancelAndErrorClosure(subscription, result));
-    }, onError: (e, st) {
-      result._completeError(e, st);
-    }, onDone: () {
-      result._complete(value);
-    }, cancelOnError: true);
+    subscription = this.listen(
+        (T element) {
+          _runUserCode(() => combine(value, element), (S newValue) {
+            value = newValue;
+          }, _cancelAndErrorClosure(subscription, result));
+        },
+        onError: result._completeError,
+        onDone: () {
+          result._complete(value);
+        },
+        cancelOnError: true);
     return result;
   }
 
   /**
-   * Collects string of data events' string representations.
+   * Combines the string representation of elements into a single string.
    *
-   * If [separator] is provided, it is inserted between any two
-   * elements.
+   * Each element is converted to a string using its [Object.toString] method.
+   * If [separator] is provided, it is inserted between element string
+   * representations.
    *
-   * Any error in the stream causes the future to complete with that
-   * error. Otherwise it completes with the collected string when
-   * the "done" event arrives.
+   * The returned future is completed with the combined string when the stream
+   * is done.
+   *
+   * If the stream contains an error, or if the call to [Object.toString]
+   * throws, the returned future is completed with that error,
+   * and procssing stops.

[floitsch, 2017/05/24 13:37:51]

processing

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

Done.

    */
   Future<String> join([String separator = ""]) {
     _Future<String> result = new _Future<String>();
     StringBuffer buffer = new StringBuffer();
     StreamSubscription subscription;
     bool first = true;
     subscription = this.listen((T element) {
       if (!first) {
         buffer.write(separator);
       }
       first = false;
       try {
         buffer.write(element);
       } catch (e, s) {
         _cancelAndErrorWithReplacement(subscription, result, e, s);
       }
     }, onError: (e) {
       result._completeError(e);
     }, onDone: () {
       result._complete(buffer.toString());
     }, cancelOnError: true);
     return result;
   }
 
   /**
    * Checks whether [needle] occurs in the elements provided by this stream.

[floitsch, 2017/05/24 13:37:50]

Returns

    *
-   * Completes the [Future] when the answer is known.
-   * If this stream reports an error, the [Future] will report that error.
+   * Compares each element of this stream to [needle] using [Object.==].
+   * If an equal element is found, the returned future is completed with `true`.
+   * If the stream ends without finding a match, the future is completed with
+   * `false`.
+   *
+   * If the stream contains an error, or the call to `Object.==` throws,
+   * the returned future is completed with that error, and processing stops.
    */
   Future<bool> contains(Object needle) {
     _Future<bool> future = new _Future<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
           _runUserCode(() => (element == needle), (bool isMatch) {
             if (isMatch) {
               _cancelAndValue(subscription, future, true);
             }
           }, _cancelAndErrorClosure(subscription, future));
         },
         onError: future._completeError,
         onDone: () {
           future._complete(false);
         },
         cancelOnError: true);
     return future;
   }
 
   /**
-   * Executes [action] on each data event of the stream.
+   * Executes [action] on each element of the stream.
    *
-   * 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.
+   * Completes the returned [Future] when all elements of the stream
+   * have been processed.
+   *
+   * If the stream contains an error, or if the call to [action] throws,
+   * the returne future completes with that error, and processing stops.
    */
   Future forEach(void action(T element)) {
     _Future future = new _Future();
     StreamSubscription subscription;
     subscription = this.listen(
         (T 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;
   }
 
   /**
    * Checks whether [test] accepts all elements provided by this stream.
    *
-   * Completes the [Future] when the answer is known.
-   * If this stream reports an error, the [Future] will report that error.
+   * Calls [test] on each element of the stream.
+   * If the call returns `false`, the returned future is completed with `false`
+   * and processing stops.
+   *
+   * If the stream ends without finding an element that [test] rejects,
+   * the returned future is completed with `true`.
+   *
+   * If this stream contains an error, or if the call to [test] throws,
+   * the returned future is completed with that error, and processing stops.
    */
   Future<bool> every(bool test(T element)) {
     _Future<bool> future = new _Future<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
           _runUserCode(() => test(element), (bool isMatch) {
             if (!isMatch) {
               _cancelAndValue(subscription, future, false);
             }
           }, _cancelAndErrorClosure(subscription, future));
         },
         onError: future._completeError,
         onDone: () {
           future._complete(true);
         },
         cancelOnError: true);
     return future;
   }
 
   /**
    * Checks whether [test] accepts any element provided by this stream.
    *
-   * Completes the [Future] when the answer is known.
+   * Calls [test] on each element of the stream.
+   * If the call returns `true`, the returned future is completed with `true`
+   * and processing stops.
    *
-   * If this stream reports an error, the [Future] reports that error.
+   * If the stream ends without finding an element that [test] accepts,
+   * the returned future is completed with `false`.
    *
-   * Stops listening to the stream after the first matching element has been
-   * found.
-   *
-   * Internally the method cancels its subscription after this element. This
-   * means that single-subscription (non-broadcast) streams are closed and
-   * cannot be reused after a call to this method.
+   * If this stream contains an error, or if the call to [test] throws,
+   * the returned future is completed with that error, and processing stops.
    */
   Future<bool> any(bool test(T element)) {
     _Future<bool> future = new _Future<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
           _runUserCode(() => test(element), (bool isMatch) {
             if (isMatch) {
               _cancelAndValue(subscription, future, true);
             }
           }, _cancelAndErrorClosure(subscription, future));
         },
         onError: future._completeError,
         onDone: () {
           future._complete(false);
         },
         cancelOnError: true);
     return future;
   }
 
-  /** Counts the elements in the stream. */
+  /**
+   * The number of elements in this stream.
+   *
+   * Waits for all elements of this stream. When the stream ends,
+   * the returned future is completed with the number of elements.
+   *
+   * If the stream contains an error, the returned future is completed with
+   * that error, and processing stops.
+   *
+   * This operation listens to the stream, and a non-broadcast stream cannot
+   * be reused after finding its length.
+   */
   Future<int> get length {
     _Future<int> future = new _Future<int>();
     int count = 0;
     this.listen(
         (_) {
           count++;
         },
         onError: future._completeError,
         onDone: () {
           future._complete(count);
         },
         cancelOnError: true);
     return future;
   }
 
   /**
-   * Reports whether this stream contains any elements.
+   * Whether this stream contains any elements.
    *
-   * Stops listening to the stream after the first element has been received.
+   * Waits for the first element of this stream, then completes the returned
+   * future with `true`.
+   * If the stream ends without emitting any elements, the returned future is
+   * completed with `false`.
    *
-   * 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.
+   * If the first event is an error, the returned future is completed with that
+   * error.
+   *
+   * This operation listens to the stream, and a non-broadcast stream cannot
+   * be reused after checking whether it is empty.
    */
   Future<bool> get isEmpty {
     _Future<bool> future = new _Future<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
         (_) {
           _cancelAndValue(subscription, future, false);
         },
         onError: future._completeError,
         onDone: () {
           future._complete(true);
         },
         cancelOnError: true);
     return future;
   }
 
-  /** Collects the data of this stream in a [List]. */
+  /**
+   * Collects all elements of this stream in a [List].
+   *
+   * Creates a `List<T>` and adds all elements of the stream to the list
+   * in the order they arrive.
+   * When the stream ends, the returned future is completed with that list.
+   *
+   * If the stream contains an error, the returned future is completed
+   * with that error, and processing stops.
+   */
   Future<List<T>> toList() {
     List<T> result = <T>[];
     _Future<List<T>> future = new _Future<List<T>>();
     this.listen(
         (T data) {
           result.add(data);
         },
         onError: future._completeError,
         onDone: () {
           future._complete(result);
@@ skipping 131 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 perform the `equals` test.
    */
   Stream<T> distinct([bool equals(T previous, T next)]) {
     return new _DistinctStream<T>(this, equals);
   }
 
   /**
-   * Returns the first element of the stream.
+   * 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.
    *
-   * If an error event occurs before the first data event, the resulting future
+   * If an error event occurs before the first data event, the returned 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].
+   * the returned future completes with an error.
    *
    * Except for the type of the error, this method is equivalent to
-   * [:this.elementAt(0):].
+   * `this.elementAt(0)`.
    */
   Future<T> get first {
     _Future<T> future = new _Future<T>();
     StreamSubscription subscription;
     subscription = this.listen(
         (T value) {
           _cancelAndValue(subscription, future, value);
         },
         onError: future._completeError,
         onDone: () {
           try {
             throw IterableElementError.noElement();
           } catch (e, s) {
             _completeWithErrorCallback(future, e, s);
           }
         },
         cancelOnError: true);
     return future;
   }
 
   /**
-   * Returns the last element of the stream.
+   * The last element of this stream.
    *
-   * If an error event occurs before the first data event, the resulting future
-   * is completed with that error.
+   * If this stream emits an error event,
+   * the returned future is completed with that error
+   * and processing stops.
    *
-   * If this stream is empty (a done event occurs before the first data event),
-   * the resulting future completes with a [StateError].
+   * If this stream is empty (the done event is the first event),
+   * the returned future completes with an error.
    */
   Future<T> get last {
     _Future<T> future = new _Future<T>();
     T result = null;
     bool foundResult = false;
     listen(
         (T value) {
           foundResult = true;
           result = value;
         },
         onError: future._completeError,
         onDone: () {
           if (foundResult) {
             future._complete(result);
             return;
           }
           try {
             throw IterableElementError.noElement();
           } catch (e, s) {
             _completeWithErrorCallback(future, e, s);
           }
         },
         cancelOnError: true);
     return future;
   }
 
   /**
-   * Returns the single element.
+   * The single element of this stream.
    *
-   * If an error event occurs before or after the first data event, the
-   * resulting future is completed with that error.
+   * If this stream emits an error event,
+   * the returned future is completed with that error
+   * and processing stops.
    *
-   * If [this] is empty or has more than one element throws a [StateError].
+   * If [this] is empty or has more than one element,
+   * the returned future completes with an error.
    */
   Future<T> get single {
     _Future<T> future = new _Future<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
         (T value) {
           if (foundResult) {
             // This is the second element we get.
@@ skipping 19 matching lines @@
             _completeWithErrorCallback(future, e, s);
           }
         },
         cancelOnError: true);
     return future;
   }
 
   /**
    * Finds the first element of this stream matching [test].
    *
-   * Returns a future that is filled with the first element of this stream
-   * that [test] returns true for.
+   * Returns a future that is completed with the first element of this stream
+   * that [test] returns `true` for.
    *
    * If no such element is found before this stream is done, and a
    * [defaultValue] function is provided, the result of calling [defaultValue]
-   * becomes the value of the future.
+   * becomes the value of the future. If [defaultValue] throws, the returned
+   * future is completed with that error.
    *
-   * Stops listening to the stream after the first matching element has been
-   * received.
+   * If this stream emits an error before the first matching element,
+   * the returned future is completed with that error, and processing stops.
+   *
+   * Stops listening to the stream after the first matching element or error
+   * has been received.
    *
    * Internally the method cancels its subscription after the first element that
    * matches the predicate. This means that single-subscription (non-broadcast)
    * streams are closed and cannot be reused after a call to this method.
    *
    * If an error occurs, or if this stream ends without finding a match and
-   * with no [defaultValue] function provided, the future will receive an
-   * error.
+   * with no [defaultValue] function provided,
+   * the returned future is completed with an error.
    */
   Future<dynamic> firstWhere(bool test(T element), {Object defaultValue()}) {
     _Future<dynamic> future = new _Future();
     StreamSubscription subscription;
     subscription = this.listen(
         (T value) {
           _runUserCode(() => test(value), (bool isMatch) {
             if (isMatch) {
               _cancelAndValue(subscription, future, value);
             }
@@ skipping 11 matching lines @@
             _completeWithErrorCallback(future, e, s);
           }
         },
         cancelOnError: true);
     return future;
   }
 
   /**
    * Finds the last element in this stream matching [test].
    *
-   * As [firstWhere], except that the last matching element is found.
-   * That means that the result cannot be provided before this stream
+   * If this stream emits an error, the returned future is completed with that
+   * error, and processing stops.
+   *
+   * Otherwise as [firstWhere], except that the last matching element is found
+   * instead of the first.
+   * That means that a non-error result cannot be provided before this stream
    * is done.
    */
   Future<dynamic> lastWhere(bool test(T element), {Object defaultValue()}) {
     _Future<dynamic> future = new _Future();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
         (T value) {
           _runUserCode(() => true == test(value), (bool isMatch) {
@@ skipping 235 matching lines @@
    * 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
    * completes with that error.
    */
   Future cancel();
 
   /**
    * Set or override the data event handler of this subscription.

[floitsch, 2017/05/24 13:37:51]

Set*s* or replaces ?
Maybe just "replaces". After all, it even replaces a `null`.

("Override" has other meanings)

Here and in the other places.

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

Just "replaces", it's cleaner!

    *
-   * This method overrides the handler that has been set at the invocation of
-   * [Stream.listen].
+   * The [handleData] function is called for each element of the stream
+   * after this function is called.
+   * If [handleData] is `null`, furhter elements are ignored.

[floitsch, 2017/05/24 13:37:51]

further

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

Done.

+   *
+   * This method overrides the current handler set by the invocation of
+   * [Stream.listen] or by a previous call to [onData].
    */
   void onData(void handleData(T data));
 
   /**
    * Set or override the error event handler of this subscription.
    *
-   * This method overrides the handler that has been set at the invocation of
-   * [Stream.listen] or by calling [asFuture].
+   * The [handleError] function must be able to be called with either
+   * one positional argument, or with two positional arguments
+   * where the seconds is always a [StackTrace].
+   *
+   * The [handleError] argument may be `null`, in which case further
+   * error events are considered unhandled, and will be reported to
+   * [Zone.handleUncaughtError].
+   *
+   * The provided function is called for all error events from the
+   * stream subscription.
+   *
+   * This method overrides the current handler set by the invocation of
+   * [Stream.listen], by calling [asFuture], or by a previous call to [onError].
    */
   void onError(Function handleError);
 
   /**
    * Set or override the done event handler of this subscription.
    *
-   * This method overrides the handler that has been set at the invocation of
-   * [Stream.listen] or by calling [asFuture].
+   * The [handleDone] function is called when the stream closes.
+   * The value may be `null`, in which case no function is called.
+   *
+   * This method overrides the current handler set by the invocation of
+   * [Stream.listen], by calling [asFuture], or by a previous call to [onDone].
    */
   void onDone(void handleDone());
 
   /**
    * Request that the stream pauses events until further notice.
    *
    * While paused, the subscription will not fire any events.
    * If it receives events from its source, they will be buffered until
    * the subscription is resumed.
-   * The underlying source is usually informed about the pause,
+   * For non-broadcast streams, the underlying source is usually informed
+   * about the pause,
    * so it can stop generating events until the subscription is resumed.
    *
    * To avoid buffering events on a broadcast stream, it is better to
    * cancel this subscription, and start to listen again when events
-   * are needed.
+   * are needed, if the intermediate events are not important.
    *
-   * If [resumeSignal] is provided, the stream will undo the pause
-   * when the future completes. If the future completes with an error,
-   * the stream will resume, but the error will not be handled!
+   * If [resumeSignal] is provided, the stream subscription will undo the pause
+   * when the future completes, as if by a call to [resume].
+   * If the future completes with an error,
+   * the stream will still resume, but the error will be considered unhandled
+   * and is passed to [Zone.handleUncaughtError].
    *
    * A call to [resume] will also undo a pause.
    *
    * If the subscription is paused more than once, an equal number
    * of resumes must be performed to resume the stream.
    *
    * Currently DOM streams silently drop events when the stream is paused. This
    * is a bug and will be fixed.
    */
   void pause([Future resumeSignal]);
 
   /**
    * Resume after a pause.
+   *
+   * This undoes one previous call to [pause].
+   * When all previously calls to [pause] has been matched by a calls to

[floitsch, 2017/05/24 13:37:51]

have been matched

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

Done.

+   * [resume], possibly through a `resumeSignal` passed to [pause],
+   * the stream subscription will again emit events.

[floitsch, 2017/05/24 13:37:50]

subscription emits again events. (No need for future tense)

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

There is no guarantee that it will emit events immediately.
To be completely pedantic, it *may* emit events again - but maybe it won't.

    */
   void resume();
 
   /**
-   * Returns true if the [StreamSubscription] is paused.
+   * Returns true if the [StreamSubscription] is currently paused.

[floitsch, 2017/05/24 13:37:51]

Whether the ...

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

Done.

+   *
+   * If there have been more calls to [pause] than to [resume] on this
+   * stream subscription, the subscription is paused, and this getter
+   * returns `true`.
+   *
+   * Returns `false` if the stream can currently emit events, or if
+   * the subscription has completed or been cancelled.
    */
   bool get isPaused;
 
   /**
    * Returns a future that handles the [onDone] and [onError] callbacks.
    *
    * 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
@@ skipping 152 matching lines @@
    * 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.
    *
    * Otherwise, the returned future will complete when either:
    *
    * * all events have been processed and the sink has been closed, or
    * * the sink has otherwise been stopped from handling more events
-   *   (for example by cancelling a stream subscription).
+   *   (for example by canceling a stream subscription).
    */
   Future get done;
 }
 
 /**
  * Transforms a Stream.
  *
  * When a stream's [Stream.transform] method is invoked with a
  * [StreamTransformer], the stream calls the [bind] method on the provided
  * transformer. The resulting stream is then returned from the
@@ skipping 201 matching lines @@
 
   void addError(error, [StackTrace stackTrace]) {
     _sink.addError(error, stackTrace);
   }
 
   void close() {
     _sink.close();
   }
 }
 
-/// A group created by [Stream.groupBy] or [Stream.groupByMapped].
+/// A group created by [Stream.groupBy].
 ///
 /// The stream created by `groupBy` emits a `GroupedEvents` for each distinct key

[floitsch, 2017/05/24 13:37:51]

long line

[Lasse Reichstein Nielsen, 2017/05/24 15:09:54]

Ack, happened with the renaming :(

 /// it encounters.
 /// This group contains the [key] itself, along with a stream of the [values]
 /// associated with that key.
 class GroupedEvents<K, V> {
   /// The key that identifiers the values emitted by [values].
   final K key;
 
   /// The [values] that [GroupBy] have grouped by the common [key].
   final Stream<V> values;
 
   factory GroupedEvents(K key, Stream<V> values) = GroupedEvents<K, V>._;
 
   // Don't expose a generative constructor.
   // This class is not intended for subclassing, so we don't want to promise
   // it. We can change that in the future.
   GroupedEvents._(this.key, this.values);
 
   /// Tells [values] to discard values instead of retaining them.
   ///
   /// Must only be used instead of listening to the [values] stream.
   /// If the stream has been listened to, this call fails.
   /// After calling this method, listening on the [values] stream fails.
   Future cancel() {
     // If values has been listened to,
     // this throws a StateError saying that stream has already been listened to,
     // which is a correct error message for this call too.
     return values.listen(null).cancel();
   }
 }