| Index: sdk/lib/async/stream_impl.dart
|
| diff --git a/sdk/lib/async/stream_impl.dart b/sdk/lib/async/stream_impl.dart
|
| index 13285a66d03d104bdb99b20fd59fb98747c89811..022955bf6db60384c7b94b99c91282905482ed4a 100644
|
| --- a/sdk/lib/async/stream_impl.dart
|
| +++ b/sdk/lib/async/stream_impl.dart
|
| @@ -442,8 +442,11 @@ class _SingleStreamImpl<T> extends _StreamImpl<T> {
|
| _subscriber = subscription;
|
| subscription._setSubscribed(0);
|
| _onSubscriptionStateChange();
|
| - // TODO(floitsch): Should this be delayed?
|
| - _handlePendingEvents();
|
| + if (_hasPendingEvent) {
|
| + new Timer(0, (_) {
|
| + _handlePendingEvents();
|
| + });
|
| + }
|
| }
|
|
|
| /**
|
| @@ -634,6 +637,58 @@ class _MultiStreamImpl<T> extends _StreamImpl<T>
|
| }
|
| }
|
|
|
| +
|
| +/** Abstract superclass for streams that generate their own events. */
|
| +abstract class _GeneratedSingleStreamImpl<T> extends _SingleStreamImpl<T> {
|
| + bool _isHandlingPendingEvents = false;
|
| + bool get _hasPendingEvent => !_isClosed;
|
| +
|
| + /**
|
| + * Generate one (or possibly more) new events.
|
| + *
|
| + * The events should be added to the stream using [_add], [_signalError] and
|
| + * [_close].
|
| + */
|
| + void _generateNextEvent();
|
| +
|
| + void _handlePendingEvents() {
|
| + // Avoid reentry from _add/_signalError/_close potentially called
|
| + // from _generateNextEvent.
|
| + if (_isHandlingPendingEvents) return;
|
| + _isHandlingPendingEvents = true;
|
| + while (!_isPaused && !_isClosed) {
|
| + // Call super's handle event in case _generateNextEvent generates
|
| + // more than one event, and the following ones are delayed.
|
| + super._handlePendingEvents();
|
| + if (!_isPaused && !_isClosed) {
|
| + _generateNextEvent();
|
| + }
|
| + }
|
| + _isHandlingPendingEvents = false;
|
| + }
|
| +}
|
| +
|
| +
|
| +/** Stream that gets its events from an [Iterable]. */
|
| +class _IterableSingleStreamImpl<T> extends _GeneratedSingleStreamImpl<T> {
|
| + Iterator<T> _iterator;
|
| +
|
| + _IterableSingleStreamImpl(Iterable<T> data) : _iterator = data.iterator;
|
| +
|
| + void _generateNextEvent() {
|
| + try {
|
| + if (_iterator.moveNext()) {
|
| + _add(_iterator.current);
|
| + return;
|
| + }
|
| + } catch (e, s) {
|
| + _signalError(new AsyncError(e, s));
|
| + }
|
| + _close();
|
| + }
|
| +}
|
| +
|
| +
|
| /**
|
| * The subscription class that the [StreamController] uses.
|
| *
|
|
|