Update documentation for "close".

sdk/lib/async/stream_impl.dart

 // Copyright (c) 2012, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 part of dart.async;
 
 /** Abstract and private interface for a place to put events. */
 abstract class _EventSink<T> {
   void _add(T data);
   void _addError(Object error, StackTrace stackTrace);
@@ skipping 711 matching lines @@
     newPrevious._next = this;
     newNext._previous = _previous;
     _previous._next = newNext;
     _previous = newPrevious;
   }
 }
 
 typedef void _broadcastCallback(StreamSubscription subscription);
 
 /**
- * Dummy subscription that will never receive any events.
+ * Done subscription that will send one done event as soon as possible.
  */
-class _DummyStreamSubscription<T> implements StreamSubscription<T> {
-  int _pauseCounter = 0;
+class _DoneStreamSubscription<T> implements StreamSubscription<T> {
+  static const int _DONE_SENT = 1;
+  static const int _SCHEDULED = 2;
+  static const int _PAUSED = 4;
+
+  final Zone _zone;
+  int _state = 0;
+  _DoneHandler _onDone;
+
+  _DoneStreamSubscription(this._onDone) : _zone = Zone.current {
+    _schedule();
+  }
+
+  bool get _isSent => (_state & _DONE_SENT) != 0;
+  bool get _isScheduled => (_state & _SCHEDULED) != 0;
+  bool get isPaused => _state >= _PAUSED;
+
+  void _schedule() {
+    if (_isScheduled) return;
+    _zone.scheduleMicrotask(_sendDone);
+    _state |= _SCHEDULED;
+  }
 
   void onData(void handleData(T data)) {}
   void onError(Function handleError) {}
-  void onDone(void handleDone()) {}
+  void onDone(void handleDone()) { _onDone = handleDone; }
 
   void pause([Future resumeSignal]) {
-    _pauseCounter++;
-    if (resumeSignal != null) resumeSignal.then((_) { resume(); });
+    _state += _PAUSED;
+    if (resumeSignal != null) resumeSignal.whenComplete(resume);
   }
+
   void resume() {
-    if (_pauseCounter > 0) _pauseCounter--;
+    if (isPaused) {
+      _state -= _PAUSED;
+      if (!isPaused && !_isSent) {
+        _schedule();
+      }
+    }
   }
+
   Future cancel() => null;
-  bool get isPaused => _pauseCounter > 0;
 
-  Future asFuture([futureValue]) => new _Future();
+  Future asFuture([futureValue]) {
+    _Future result = new _Future();
+    _onDone = () { result._completeWithValue(null); }
+    return result;
+  }
+
+  void _sendDone() {
+    _state &= ~_SCHEDULED;
+    if (isPaused) return;
+    _state |= _DONE_SENT;
+    _zone.runGuarded(_onDone);
+  }
 }
 
 class _AsBroadcastStream<T> extends Stream<T> {
   final Stream<T> _source;
   final _broadcastCallback _onListenHandler;
   final _broadcastCallback _onCancelHandler;
   final Zone _zone;
 
   _AsBroadcastStreamController<T> _controller;
   StreamSubscription<T> _subscription;
 
   _AsBroadcastStream(this._source,
                      void onListenHandler(StreamSubscription subscription),
                      void onCancelHandler(StreamSubscription subscription))
       : _onListenHandler = Zone.current.registerUnaryCallback(onListenHandler),
         _onCancelHandler = Zone.current.registerUnaryCallback(onCancelHandler),
         _zone = Zone.current {
     _controller = new _AsBroadcastStreamController<T>(_onListen, _onCancel);
   }
 
   bool get isBroadcast => true;
 
   StreamSubscription<T> listen(void onData(T data),
                                { Function onError,
                                  void onDone(),
                                  bool cancelOnError}) {
-    if (_controller == null) {
+    if (_controller == null || _controller.isClosed) {
       // Return a dummy subscription backed by nothing, since
-      // it won't ever receive any events.
-      return new _DummyStreamSubscription<T>();
+      // it will only ever send one done event.
+      return new _DoneStreamSubscription<T>(onDone);
     }
     if (_subscription == null) {
       _subscription = _source.listen(_controller.add,
                                      onError: _controller.addError,
                                      onDone: _controller.close);
     }
     cancelOnError = identical(true, cancelOnError);
     StreamSubscription<T> result = _controller._subscribe(cancelOnError);
     result.onData(onData);
     result.onError(onError);
@@ skipping 237 matching lines @@
       _Future<bool> hasNext = _futureOrPrefetch;
       _clear();
       hasNext._complete(false);
       return;
     }
     _subscription.pause();
     _futureOrPrefetch = null;
     _state = _STATE_EXTRA_DONE;
   }
 }