Change and structure how Stream implementations do callbacks.

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
 // -------------------------------------------------------------------
 
 /**
  * A source of asynchronous data events.
  *
  * A Stream provides a sequence of events. Each event is either a data event or
  * an error event, representing the result of a single computation. When the
  * Stream is exhausted, it may send a single "done" event.
  *
  * You can [listen] on a stream to receive the events it sends. When you listen,
  * you receive a [StreamSubscription] object that can be used to stop listening,
  * or to temporarily pause events from the stream.
  *
  * When an event is fired, the listeners at that time are informed.
- * If a listener is added or removed while an event is being fired, the change
- * will only take effect after the event is completely fired.
+ * If a listener is added while an event is being fired, the change
+ * will only take effect after the event is completely 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 after this point is allowed, but
+ * they will just receive a new "done" event as soon as possible.
  *
  * Streams 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.
  *
  * There are two kinds of streams: The normal "single-subscription" streams and
  * "broadcast" streams.
  *
  * A single-subscription stream allows only a single listener at a time.
  * It holds back events until it gets a listener, and it may exhaust
@@ skipping 1071 matching lines @@
 }
 
 class _StreamOutputSinkWrapper<T> implements StreamSink<T> {
   _StreamOutputSink _sink;
   _StreamOutputSinkWrapper(this._sink);
 
   void add(T data) => _sink._sendData(data);
   void signalError(AsyncError error) => _sink._sendError(error);
   void close() => _sink._sendDone();
 }