| Index: sdk/lib/async/stream_impl.dart
|
| diff --git a/sdk/lib/async/stream_impl.dart b/sdk/lib/async/stream_impl.dart
|
| index 78c408772e60387f6a3289276d61dadfedd3cdbf..2fd95fde362d55cd6e9b207bc139d89b7f5f9042 100644
|
| --- a/sdk/lib/async/stream_impl.dart
|
| +++ b/sdk/lib/async/stream_impl.dart
|
| @@ -729,26 +729,63 @@ class _BroadcastLinkedList {
|
| 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> {
|
| @@ -775,10 +812,10 @@ class _AsBroadcastStream<T> extends Stream<T> {
|
| { 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,
|
|
|