From df2bcb893f941ce93ddf7819b39c5fa342ee2fd6 Mon Sep 17 00:00:00 2001 From: "floitsch@google.com" Date: Fri, 28 Jun 2013 15:07:09 +0000 Subject: [PATCH] Revert "Make StreamController be a StreamSink, not just an EventSink." Revert "Remove type variable to work around dart2js bug." This reverts commit r24587. This reverts commit r24588. Review URL: https://codereview.chromium.org//18080015 git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@24590 260f80e4-7a28-3924-810f-c04153c831b5 --- sdk/lib/async/async.dart | 1 - sdk/lib/async/async_sources.gypi | 1 - .../async/broadcast_stream_controller.dart | 492 ----------- sdk/lib/async/future_impl.dart | 47 +- sdk/lib/async/stream.dart | 13 + sdk/lib/async/stream_controller.dart | 816 ++++++++++-------- sdk/lib/async/stream_impl.dart | 7 +- tests/lib/async/event_helper.dart | 8 - .../async/stream_controller_async_test.dart | 115 +-- tests/lib/async/stream_controller_test.dart | 4 +- 10 files changed, 478 insertions(+), 1026 deletions(-) delete mode 100644 sdk/lib/async/broadcast_stream_controller.dart diff --git a/sdk/lib/async/async.dart b/sdk/lib/async/async.dart index 5f843b2e409..acd808aaf99 100644 --- a/sdk/lib/async/async.dart +++ b/sdk/lib/async/async.dart @@ -7,7 +7,6 @@ library dart.async; import "dart:collection"; part 'async_error.dart'; -part 'broadcast_stream_controller.dart'; part 'deferred_load.dart'; part 'event_loop.dart'; part 'future.dart'; diff --git a/sdk/lib/async/async_sources.gypi b/sdk/lib/async/async_sources.gypi index 3f8d32990f8..dc0dbecaa1f 100644 --- a/sdk/lib/async/async_sources.gypi +++ b/sdk/lib/async/async_sources.gypi @@ -8,7 +8,6 @@ 'async.dart', # The above file needs to be first as it lists the parts below. 'async_error.dart', - 'broadcast_stream_controller.dart', 'deferred_load.dart', 'event_loop.dart', 'future.dart', diff --git a/sdk/lib/async/broadcast_stream_controller.dart b/sdk/lib/async/broadcast_stream_controller.dart deleted file mode 100644 index 8f4b019b315..00000000000 --- a/sdk/lib/async/broadcast_stream_controller.dart +++ /dev/null @@ -1,492 +0,0 @@ -// 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; - -class _BroadcastStream extends _StreamImpl { - _BroadcastStreamController _controller; - - _BroadcastStream(this._controller); - - bool get isBroadcast => true; - - StreamSubscription _createSubscription( - void onData(T data), - void onError(Object error), - void onDone(), - bool cancelOnError) => - _controller._subscribe(onData, onError, onDone, cancelOnError); -} - -abstract class _BroadcastSubscriptionLink { - _BroadcastSubscriptionLink _next; - _BroadcastSubscriptionLink _previous; -} - -class _BroadcastSubscription extends _ControllerSubscription - implements _BroadcastSubscriptionLink { - static const int _STATE_EVENT_ID = 1; - static const int _STATE_FIRING = 2; - static const int _STATE_REMOVE_AFTER_FIRING = 4; - // TODO(lrn): Use the _state field on _ControllerSubscription to - // also store this state. Requires that the subscription implementation - // does not assume that it's use of the state integer is the only use. - int _eventState; - - _BroadcastSubscriptionLink _next; - _BroadcastSubscriptionLink _previous; - - _BroadcastSubscription(_StreamControllerLifecycle controller, - void onData(T data), - void onError(Object error), - void onDone(), - bool cancelOnError) - : super(controller, onData, onError, onDone, cancelOnError) { - _next = _previous = this; - } - - _BroadcastStreamController get _controller => super._controller; - - bool _expectsEvent(int eventId) => - (_eventState & _STATE_EVENT_ID) == eventId; - - - void _toggleEventId() { - _eventState ^= _STATE_EVENT_ID; - } - - bool get _isFiring => (_eventState & _STATE_FIRING) != 0; - - bool _setRemoveAfterFiring() { - assert(_isFiring); - _eventState |= _STATE_REMOVE_AFTER_FIRING; - } - - bool get _removeAfterFiring => - (_eventState & _STATE_REMOVE_AFTER_FIRING) != 0; - - // The controller._recordPause doesn't do anything for a broadcast controller, - // so we don't bother calling it. - void _onPause() { } - - // The controller._recordResume doesn't do anything for a broadcast - // controller, so we don't bother calling it. - void _onResume() { } - - // _onCancel is inherited. -} - - -abstract class _BroadcastStreamController - implements StreamController, - _StreamControllerLifecycle, - _BroadcastSubscriptionLink, - _EventSink, - _EventDispatch { - static const int _STATE_INITIAL = 0; - static const int _STATE_EVENT_ID = 1; - static const int _STATE_FIRING = 2; - static const int _STATE_CLOSED = 4; - static const int _STATE_ADDSTREAM = 8; - - final _NotificationHandler _onListen; - final _NotificationHandler _onCancel; - - // State of the controller. - int _state; - - // Double-linked list of active listeners. - _BroadcastSubscriptionLink _next; - _BroadcastSubscriptionLink _previous; - - // Extra state used during an [addStream] call. - _AddStreamState _addStreamState; - - /** - * Future returned by [close] and [done]. - * - * The future is completed whenever the done event has been sent to all - * relevant listeners. - * The relevant listeners are the ones that were listening when [close] was - * called. When all of these have been canceled (sending the done event makes - * them cancel, but they can also be canceled before sending the event), - * this future completes. - * - * Any attempt to listen after calling [close] will throw, so there won't - * be any further listeners. - */ - _FutureImpl _doneFuture; - - _BroadcastStreamController(this._onListen, this._onCancel) - : _state = _STATE_INITIAL { - _next = _previous = this; - } - - // StreamController interface. - - Stream get stream => new _BroadcastStream(this); - - StreamSink get sink => new _StreamSinkWrapper(this); - - bool get isClosed => (_state & _STATE_CLOSED) != 0; - - /** - * A broadcast controller is never paused. - * - * Each receiving stream may be paused individually, and they handle their - * own buffering. - */ - bool get isPaused => false; - - /** Whether there are currently one or more subscribers. */ - bool get hasListener => !_isEmpty; - - /** Whether an event is being fired (sent to some, but not all, listeners). */ - bool get _isFiring => (_state & _STATE_FIRING) != 0; - - bool get _isAddingStream => (_state & _STATE_ADDSTREAM) != 0; - - bool get _mayAddEvent => (_state < _STATE_CLOSED); - - _FutureImpl _ensureDoneFuture() { - if (_doneFuture != null) return _doneFuture; - return _doneFuture = new _FutureImpl(); - } - - // Linked list helpers - - bool get _isEmpty => identical(_next, this); - - /** Adds subscription to linked list of active listeners. */ - void _addListener(_BroadcastSubscription subscription) { - assert(identical(subscription._next, subscription)); - // Insert in linked list just before `this`. - subscription._previous = _previous; - subscription._next = this; - this._previous._next = subscription; - this._previous = subscription; - subscription._eventState = (_state & _STATE_EVENT_ID); - } - - void _removeListener(_BroadcastSubscription subscription) { - assert(identical(subscription._controller, this)); - assert(!identical(subscription._next, subscription)); - _BroadcastSubscriptionLink previous = subscription._previous; - _BroadcastSubscriptionLink next = subscription._next; - previous._next = next; - next._previous = previous; - subscription._next = subscription._previous = subscription; - } - - // _StreamControllerLifecycle interface. - - StreamSubscription _subscribe(void onData(T data), - void onError(Object error), - void onDone(), - bool cancelOnError) { - if (isClosed) { - throw new StateError("Subscribing to closed stream"); - } - StreamSubscription subscription = new _BroadcastSubscription( - this, onData, onError, onDone, cancelOnError); - _addListener(subscription); - if (identical(_next, _previous)) { - // Only one listener, so it must be the first listener. - _runGuarded(_onListen); - } - return subscription; - } - - void _recordCancel(_BroadcastSubscription subscription) { - // If already removed by the stream, don't remove it again. - if (identical(subscription._next, subscription)) return; - assert(!identical(subscription._next, subscription)); - if (subscription._isFiring) { - subscription._setRemoveAfterFiring(); - } else { - assert(!identical(subscription._next, subscription)); - _removeListener(subscription); - // If we are currently firing an event, the empty-check is performed at - // the end of the listener loop instead of here. - if (!_isFiring && _isEmpty) { - _callOnCancel(); - } - } - } - - void _recordPause(StreamSubscription subscription) {} - void _recordResume(StreamSubscription subscription) {} - - // EventSink interface. - - Error _addEventError() { - if (isClosed) { - return new StateError("Cannot add new events after calling close"); - } - assert(_isAddingStream); - return new StateError("Cannot add new events while doing an addStream"); - } - - void add(T data) { - if (!_mayAddEvent) throw _addEventError(); - _sendData(data); - } - - void addError(Object error, [Object stackTrace]) { - if (!_mayAddEvent) throw _addEventError(); - if (stackTrace != null) _attachStackTrace(error, stackTrace); - _sendError(error); - } - - Future close() { - if (isClosed) { - assert(_doneFuture != null); - return _doneFuture; - } - if (!_mayAddEvent) throw _addEventError(); - _state |= _STATE_CLOSED; - Future doneFuture = _ensureDoneFuture(); - _sendDone(); - return doneFuture; - } - - Future get done => _ensureDoneFuture(); - - Future addStream(Stream stream) { - if (!_mayAddEvent) throw _addEventError(); - _state |= _STATE_ADDSTREAM; - _addStreamState = new _AddStreamState(this, stream); - return _addStreamState.addStreamFuture; - } - - // _EventSink interface, called from AddStreamState. - void _add(T data) { - _sendData(data); - } - - void _addError(Object error) { - assert(_isAddingStream); - _sendError(error); - } - - void _close() { - assert(_isAddingStream); - _AddStreamState addState = _addStreamState; - _addStreamState = null; - _state &= ~_STATE_ADDSTREAM; - addState.complete(); - } - - // Event handling. - void _forEachListener( - void action(_BufferingStreamSubscription subscription)) { - if (_isFiring) { - throw new StateError( - "Cannot fire new event. Controller is already firing an event"); - } - if (_isEmpty) return; - - // Get event id of this event. - int id = (_state & _STATE_EVENT_ID); - // Start firing (set the _STATE_FIRING bit). We don't do [_onCancel] - // callbacks while firing, and we prevent reentrancy of this function. - // - // Set [_state]'s event id to the next event's id. - // Any listeners added while firing this event will expect the next event, - // not this one, and won't get notified. - _state ^= _STATE_EVENT_ID | _STATE_FIRING; - _BroadcastSubscriptionLink link = _next; - while (!identical(link, this)) { - _BroadcastSubscription subscription = link; - if (subscription._expectsEvent(id)) { - subscription._eventState |= _BroadcastSubscription._STATE_FIRING; - action(subscription); - subscription._toggleEventId(); - link = subscription._next; - if (subscription._removeAfterFiring) { - _removeListener(subscription); - } - subscription._eventState &= ~_BroadcastSubscription._STATE_FIRING; - } else { - link = subscription._next; - } - } - _state &= ~_STATE_FIRING; - - if (_isEmpty) { - _callOnCancel(); - } - } - - void _callOnCancel() { - assert(_isEmpty); - if (isClosed && _doneFuture._mayComplete) { - // When closed, _doneFuture is not null. - _doneFuture._asyncSetValue(null); - } - _runGuarded(_onCancel); - } -} - -class _SyncBroadcastStreamController extends _BroadcastStreamController { - _SyncBroadcastStreamController(void onListen(), void onCancel()) - : super(onListen, onCancel); - - // EventDispatch interface. - - void _sendData(T data) { - if (_isEmpty) return; - _forEachListener((_BufferingStreamSubscription subscription) { - subscription._add(data); - }); - } - - void _sendError(Object error) { - if (_isEmpty) return; - _forEachListener((_BufferingStreamSubscription subscription) { - subscription._addError(error); - }); - } - - void _sendDone() { - if (!_isEmpty) { - _forEachListener((_BroadcastSubscription subscription) { - subscription._close(); - }); - } else { - assert(_doneFuture != null); - assert(_doneFuture._mayComplete); - _doneFuture._asyncSetValue(null); - } - } -} - -class _AsyncBroadcastStreamController extends _BroadcastStreamController { - _AsyncBroadcastStreamController(void onListen(), void onCancel()) - : super(onListen, onCancel); - - // EventDispatch interface. - - void _sendData(T data) { - for (_BroadcastSubscriptionLink link = _next; - !identical(link, this); - link = link._next) { - _BroadcastSubscription subscription = link; - subscription._addPending(new _DelayedData(data)); - } - } - - void _sendError(Object error) { - for (_BroadcastSubscriptionLink link = _next; - !identical(link, this); - link = link._next) { - _BroadcastSubscription subscription = link; - subscription._addPending(new _DelayedError(error)); - } - } - - void _sendDone() { - if (!_isEmpty) { - for (_BroadcastSubscriptionLink link = _next; - !identical(link, this); - link = link._next) { - _BroadcastSubscription subscription = link; - subscription._addPending(const _DelayedDone()); - } - } else { - assert(_doneFuture != null); - assert(_doneFuture._mayComplete); - _doneFuture._asyncSetValue(null); - } - } -} - -/** - * Stream controller that is used by [Stream.asBroadcastStream]. - * - * This stream controller allows incoming events while it is firing - * other events. This is handled by delaying the events until the - * current event is done firing, and then fire the pending events. - * - * This class extends [_SyncBroadcastStreamController]. Events of - * an "asBroadcastStream" stream are always initiated by events - * on another stream, and it is fine to forward them synchronously. - */ -class _AsBroadcastStreamController - extends _SyncBroadcastStreamController - implements _EventDispatch { - _StreamImplEvents _pending; - - _AsBroadcastStreamController(void onListen(), void onCancel()) - : super(onListen, onCancel); - - bool get _hasPending => _pending != null && ! _pending.isEmpty; - - void _addPendingEvent(_DelayedEvent event) { - if (_pending == null) { - _pending = new _StreamImplEvents(); - } - _pending.add(event); - } - - void add(T data) { - if (!isClosed && _isFiring) { - _addPendingEvent(new _DelayedData(data)); - return; - } - super.add(data); - while (_hasPending) { - _pending.handleNext(this); - } - } - - void addError(Object error, [StackTrace stackTrace]) { - if (!isClosed && _isFiring) { - _addPendingEvent(new _DelayedError(error)); - return; - } - super.addError(error, stackTrace); - while (_hasPending) { - _pending.handleNext(this); - } - } - - void close() { - if (!isClosed && _isFiring) { - _addPendingEvent(const _DelayedDone()); - _state |= _STATE_CLOSED; - return; - } - super.close(); - assert(!_hasPending); - } - - void _callOnCancel() { - if (_hasPending) { - _pending.clear(); - _pending = null; - } - super._callOnCancel(); - } -} - -// A subscription that never receives any events. -// It can simulate pauses, but otherwise does nothing. -class _DoneSubscription implements StreamSubscription { - int _pauseCount = 0; - void onData(void handleData(T data)) {} - void onError(void handleErrr(Object error)) {} - void onDone(void handleDone()) {} - void pause([Future resumeSignal]) { - if (resumeSignal != null) resumeSignal.then(_resume); - _pauseCount++; - } - void resume() { _resume(null); } - void _resume(_) { - if (_pauseCount > 0) _pauseCount--; - } - void cancel() {} - bool get isPaused => _pauseCount > 0; - Future asFuture(Object value) => new _FutureImpl(); -} diff --git a/sdk/lib/async/future_impl.dart b/sdk/lib/async/future_impl.dart index 73dc35f18cf..be0b7a23229 100644 --- a/sdk/lib/async/future_impl.dart +++ b/sdk/lib/async/future_impl.dart @@ -49,12 +49,12 @@ abstract class _Completer implements Completer { class _AsyncCompleter extends _Completer { void _setFutureValue(T value) { _FutureImpl future = this.future; - future._asyncSetValue(value); + runAsync(() { future._setValue(value); }); } void _setFutureError(error) { _FutureImpl future = this.future; - future._asyncSetError(error); + runAsync(() { future._setError(error); }); } } @@ -94,8 +94,8 @@ class _FutureListenerWrapper implements _FutureListener { _FutureImpl future; _FutureListener _nextListener; _FutureListenerWrapper(this.future); - _sendValue(T value) { future._setValueUnchecked(value); } - _sendError(error) { future._setErrorUnchecked(error); } + _sendValue(T value) { future._setValue(value); } + _sendError(error) { future._setError(error); } bool _inSameErrorZone(_Zone otherZone) => future._inSameErrorZone(otherZone); } @@ -162,29 +162,26 @@ class _FutureImpl implements Future { /// [resultOrListeners] field holds a single-linked list of /// [FutureListener] listeners. static const int _INCOMPLETE = 0; - /// Pending completion. Set when completed using [_asyncSetValue] or - /// [_asyncSetError]. It is an error to try to complete it again. - static const int _PENDING_COMPLETE = 1; /// The future has been chained to another future. The result of that /// other future becomes the result of this future as well. /// In this state, the [resultOrListeners] field holds the future that /// will give the result to this future. Both existing and new listeners are /// forwarded directly to the other future. - static const int _CHAINED = 2; + static const int _CHAINED = 1; /// The future has been chained to another future, but there hasn't been /// any listeners added to this future yet. If it is completed with an /// error, the error will be considered unhandled. - static const int _CHAINED_UNLISTENED = 6; + static const int _CHAINED_UNLISTENED = 3; /// The future has been completed with a value result. - static const int _VALUE = 8; + static const int _VALUE = 4; /// The future has been completed with an error result. - static const int _ERROR = 12; + static const int _ERROR = 6; /// Extra bit set when the future has been completed with an error result. /// but no listener has been scheduled to receive the error. /// If the bit is still set when a [runAsync] call triggers, the error will /// be reported to the top-level handler. /// Assigning a listener before that time will clear the bit. - static const int _UNHANDLED_ERROR = 16; + static const int _UNHANDLED_ERROR = 8; /** Whether the future is complete, and as what. */ int _state = _INCOMPLETE; @@ -194,7 +191,6 @@ class _FutureImpl implements Future { bool get _isChained => (_state & _CHAINED) != 0; bool get _hasChainedListener => _state == _CHAINED; bool get _isComplete => _state >= _VALUE; - bool get _mayComplete => _state == _INCOMPLETE; bool get _hasValue => _state == _VALUE; bool get _hasError => _state >= _ERROR; bool get _hasUnhandledError => _state >= _UNHANDLED_ERROR; @@ -294,11 +290,7 @@ class _FutureImpl implements Future { } void _setValue(T value) { - if (!_mayComplete) throw new StateError("Future already completed"); - _setValueUnchecked(value); - } - - void _setValueUnchecked(T value) { + if (_isComplete) throw new StateError("Future already completed"); _FutureListener listeners = _isChained ? null : _removeListeners(); _state = _VALUE; _resultOrListeners = value; @@ -310,12 +302,9 @@ class _FutureImpl implements Future { } } - void _setError(Object error) { - if (!_mayComplete) throw new StateError("Future already completed"); - _setErrorUnchecked(error); - } + void _setError(error) { + if (_isComplete) throw new StateError("Future already completed"); - void _setErrorUnchecked(Object error) { _FutureListener listeners; bool hasListeners; if (_isChained) { @@ -341,18 +330,6 @@ class _FutureImpl implements Future { } } - void _asyncSetValue(T value) { - if (!_mayComplete) throw new StateError("Future already completed"); - _state = _PENDING_COMPLETE; - runAsync(() { _setValueUnchecked(value); }); - } - - void _asyncSetError(Object error) { - if (!_mayComplete) throw new StateError("Future already completed"); - _state = _PENDING_COMPLETE; - runAsync(() { _setErrorUnchecked(error); }); - } - void _scheduleUnhandledError() { assert(_state == _ERROR); _state = _ERROR | _UNHANDLED_ERROR; diff --git a/sdk/lib/async/stream.dart b/sdk/lib/async/stream.dart index 0a0208ba718..0de29194a51 100644 --- a/sdk/lib/async/stream.dart +++ b/sdk/lib/async/stream.dart @@ -925,6 +925,19 @@ class StreamView extends Stream { } } +/** + * [EventSink] wrapper that only exposes the [EventSink] interface. + */ +class _EventSinkView extends EventSink { + final EventSink _sink; + + _EventSinkView(this._sink); + + void add(T value) { _sink.add(value); } + void addError(error) { _sink.addError(error); } + void close() { _sink.close(); } +} + /** * The target of a [Stream.pipe] call. diff --git a/sdk/lib/async/stream_controller.dart b/sdk/lib/async/stream_controller.dart index 3ab9c47b273..2cdbeaf240c 100644 --- a/sdk/lib/async/stream_controller.dart +++ b/sdk/lib/async/stream_controller.dart @@ -46,7 +46,7 @@ part of dart.async; * the stream at all, and won't trigger callbacks. From the controller's point * of view, the stream is completely inert when has completed. */ -abstract class StreamController implements StreamSink { +abstract class StreamController implements EventSink { /** The stream that this controller is controlling. */ Stream get stream; @@ -75,17 +75,10 @@ abstract class StreamController implements StreamSink { void onPause(), void onResume(), void onCancel(), - bool sync: false}) { - if (onListen == null && onPause == null && - onResume == null && onCancel == null) { - return sync - ? new _NoCallbackSyncStreamController/**/() - : new _NoCallbackAsyncStreamController/**/(); - } - return sync + bool sync: false}) + => sync ? new _SyncStreamController(onListen, onPause, onResume, onCancel) : new _AsyncStreamController(onListen, onPause, onResume, onCancel); - } /** * A controller where [stream] can be listened to more than once. @@ -130,9 +123,9 @@ abstract class StreamController implements StreamSink { } /** - * Returns a view of this object that only exposes the [StreamSink] interface. + * Returns a view of this object that only exposes the [EventSink] interface. */ - StreamSink get sink; + EventSink get sink; /** * Whether the stream is closed for adding more events. @@ -169,10 +162,7 @@ abstract class StreamController implements StreamSink { abstract class _StreamControllerLifecycle { - StreamSubscription _subscribe(void onData(T data), - void onError(Object error), - void onDone(), - bool cancelOnError); + void _recordListen(StreamSubscription subscription) {} void _recordPause(StreamSubscription subscription) {} void _recordResume(StreamSubscription subscription) {} void _recordCancel(StreamSubscription subscription) {} @@ -185,205 +175,81 @@ abstract class _StreamControllerLifecycle { */ abstract class _StreamController implements StreamController, _StreamControllerLifecycle, - _EventSink, _EventDispatch { - // The states are bit-flags. More than one can be set at a time. - // - // The "subscription state" goes through the states: - // initial -> subscribed -> canceled. - // These are mutually exclusive. - // The "closed" state records whether the [close] method has been called - // on the controller. This can be done at any time. If done before - // subscription, the done event is queued. If done after cancel, the done - // event is ignored (just as any other event after a cancel). + static const int _STATE_OPEN = 0; + static const int _STATE_CANCELLED = 1; + static const int _STATE_CLOSED = 2; - /** The controller is in its initial state with no subscription. */ - static const int _STATE_INITIAL = 0; - /** The controller has a subscription, but hasn't been closed or canceled. */ - static const int _STATE_SUBSCRIBED = 1; - /** The subscription is canceled. */ - static const int _STATE_CANCELED = 2; - /** Mask for the subscription state. */ - static const int _STATE_SUBSCRIPTION_MASK = 3; + final _NotificationHandler _onListen; + final _NotificationHandler _onPause; + final _NotificationHandler _onResume; + final _NotificationHandler _onCancel; + _StreamImpl _stream; - // The following state relate to the controller, not the subscription. - // If closed, adding more events is not allowed. - // If executing an [addStream], new events are not allowed either, but will - // be added by the stream. + // An active subscription on the stream, or null if no subscripton is active. + _ControllerSubscription _subscription; + + // Whether we have sent a "done" event. + int _state = _STATE_OPEN; + + // Events added to the stream before it has an active subscription. + _PendingEvents _pendingEvents = null; + + _StreamController(this._onListen, + this._onPause, + this._onResume, + this._onCancel) { + _stream = new _ControllerStream(this); + } + + Stream get stream => _stream; /** - * The controller is closed due to calling [close]. - * - * When the stream is closed, you can neither add new events nor add new - * listeners. + * Returns a view of this object that only exposes the [EventSink] interface. */ - static const int _STATE_CLOSED = 4; - /** - * The controller is in the middle of an [addStream] operation. - * - * While adding events from a stream, no new events can be added directly - * on the controller. - */ - static const int _STATE_ADDSTREAM = 8; + EventSink get sink => new _EventSinkView(this); /** - * Field containing different data depending on the current subscription - * state. - * - * If [_state] is [_STATE_INITIAL], the field may contain a [_PendingEvents] - * for events added to the controller before a subscription. - * - * While [_state] is [_STATE_SUBSCRIBED], the field contains the subscription. - * - * When [_state] is [_STATE_CANCELED] the field is currently not used. - */ - var _varData; - - /** Current state of the controller. */ - int _state = _STATE_INITIAL; - - /** - * Future completed when the stream sends its last event. - * - * This is also the future returned by [close]. - */ - // TODO(lrn): Could this be stored in the varData field too, if it's not - // accessed until the call to "close"? Then we need to special case if it's - // accessed earlier, or if close is called before subscribing. - _FutureImpl _doneFuture; - - _StreamController(); - - _NotificationHandler get _onListen; - _NotificationHandler get _onPause; - _NotificationHandler get _onResume; - _NotificationHandler get _onCancel; - - // Return a new stream every time. The streams are equal, but not identical. - Stream get stream => new _ControllerStream(this); - - /** - * Returns a view of this object that only exposes the [StreamSink] interface. - */ - StreamSink get sink => new _StreamSinkWrapper(this); - - /** - * Whether a listener has existed and been canceled. + * Whether a listener has existed and been cancelled. * * After this, adding more events will be ignored. */ - bool get _isCanceled => (_state & _STATE_CANCELED) != 0; - - /** Whether there is an active listener. */ - bool get hasListener => (_state & _STATE_SUBSCRIBED) != 0; - - /** Whether there has not been a listener yet. */ - bool get _isInitialState => - (_state & _STATE_SUBSCRIPTION_MASK) == _STATE_INITIAL; + bool get _isCancelled => (_state & _STATE_CANCELLED) != 0; bool get isClosed => (_state & _STATE_CLOSED) != 0; bool get isPaused => hasListener ? _subscription._isInputPaused - : !_isCanceled; + : !_isCancelled; - bool get _isAddingStream => (_state & _STATE_ADDSTREAM) != 0; - - /** New events may not be added after close, or during addStream. */ - bool get _mayAddEvent => (_state < _STATE_CLOSED); - - // Returns the pending events. - // Pending events are events added before a subscription exists. - // They are added to the subscription when it is created. - // Pending events, if any, are kept in the _varData field until the - // stream is listened to. - // While adding a stream, pending events are moved into the - // state object to allow the state object to use the _varData field. - _PendingEvents get _pendingEvents { - assert(_isInitialState); - if (!_isAddingStream) { - return _varData; - } - _StreamControllerAddStreamState state = _varData; - return state.varData; - } - - // Returns the pending events, and creates the object if necessary. - _StreamImplEvents _ensurePendingEvents() { - assert(_isInitialState); - if (!_isAddingStream) { - if (_varData == null) _varData = new _StreamImplEvents(); - return _varData; - } - _StreamControllerAddStreamState state = _varData; - if (state.varData == null) state.varData = new _StreamImplEvents(); - return state.varData; - } - - // Get the current subscription. - // If we are adding a stream, the subscription is moved into the state - // object to allow the state object to use the _varData field. - _ControllerSubscription get _subscription { - assert(hasListener); - if (_isAddingStream) { - _StreamControllerAddStreamState addState = _varData; - return addState.varData; - } - return _varData; - } + bool get hasListener => _subscription != null; /** - * Creates an error describing why an event cannot be added. - * - * The reason, and therefore the error message, depends on the current state. - */ - Error _badEventState() { - if (isClosed) { - return new StateError("Cannot add event after closing"); - } - assert(_isAddingStream); - return new StateError("Cannot add event while adding a stream"); - } - - // StreamSink interface. - Future addStream(Stream source) { - if (!_mayAddEvent) throw _badEventState(); - if (_isCanceled) return new _FutureImpl.immediate(null); - _StreamControllerAddStreamState addState = - new _StreamControllerAddStreamState(this, _varData, source); - _varData = addState; - _state |= _STATE_ADDSTREAM; - return addState.addStreamFuture; - } - - Future get done => _ensureDoneFuture(); - - Future _ensureDoneFuture() { - if (_doneFuture == null) { - _doneFuture = new _FutureImpl(); - if (_isCanceled) _doneFuture._setValue(null); - } - return _doneFuture; - } - - /** - * Send or enqueue a data event. + * Send or queue a data event. */ void add(T value) { - if (!_mayAddEvent) throw _badEventState(); - _add(value); + if (isClosed) throw new StateError("Adding event after close"); + if (_subscription != null) { + _sendData(value); + } else if (!_isCancelled) { + _addPendingEvent(new _DelayedData(value)); + } } /** * Send or enqueue an error event. */ void addError(Object error, [Object stackTrace]) { - if (!_mayAddEvent) throw _badEventState(); + if (isClosed) throw new StateError("Adding event after close"); if (stackTrace != null) { // Force stack trace overwrite. Even if the error already contained // a stack trace. _attachStackTrace(error, stackTrace); } - _addError(error); + if (_subscription != null) { + _sendError(error); + } else if (!_isCancelled) { + _addPendingEvent(new _DelayedError(error)); + } } /** @@ -397,111 +263,60 @@ abstract class _StreamController implements StreamController, * The first time a controller is closed, a "done" event is sent to its * stream. */ - Future close() { - if (isClosed) { - assert(_doneFuture != null); // Was set when close was first called. - return _doneFuture; - } - if (!_mayAddEvent) throw _badEventState(); + void close() { + if (isClosed) return; _state |= _STATE_CLOSED; - _ensureDoneFuture(); - if (hasListener) { + if (_subscription != null) { _sendDone(); - } else if (_isInitialState) { - _ensurePendingEvents().add(const _DelayedDone()); - } - return _doneFuture; - } - - // EventSink interface. Used by the [addStream] events. - - // Add data event, used both by the [addStream] events and by [add]. - void _add(T value) { - if (hasListener) { - _sendData(value); - } else if (_isInitialState) { - _ensurePendingEvents().add(new _DelayedData(value)); + } else if (!_isCancelled) { + _addPendingEvent(const _DelayedDone()); } } - void _addError(Object error) { - if (hasListener) { - _sendError(error); - } else if (_isInitialState) { - _ensurePendingEvents().add(new _DelayedError(error)); + // EventDispatch interface + + void _addPendingEvent(_DelayedEvent event) { + if (_isCancelled) return; + _StreamImplEvents events = _pendingEvents; + if (events == null) { + events = _pendingEvents = new _StreamImplEvents(); } + events.add(event); } - void _close() { - // End of addStream stream. - assert(_isAddingStream); - _StreamControllerAddStreamState addState = _varData; - _varData = addState.varData; - _state &= ~_STATE_ADDSTREAM; - addState.complete(); - } - - // _StreamControllerLifeCycle interface - - StreamSubscription _subscribe(void onData(T data), - void onError(Object error), - void onDone(), - bool cancelOnError) { - if (!_isInitialState) { - throw new StateError("Stream has already been listened to."); - } - _ControllerSubscription subscription = new _ControllerSubscription( - this, onData, onError, onDone, cancelOnError); - - _PendingEvents pendingEvents = _pendingEvents; - _state |= _STATE_SUBSCRIBED; - if (_isAddingStream) { - _StreamControllerAddStreamState addState = _varData; - addState.varData = subscription; - } else { - _varData = subscription; - } - subscription._setPendingEvents(pendingEvents); + void _recordListen(_BufferingStreamSubscription subscription) { + assert(_subscription == null); + _subscription = subscription; + subscription._setPendingEvents(_pendingEvents); + _pendingEvents = null; subscription._guardCallback(() { _runGuarded(_onListen); }); - - return subscription; } void _recordCancel(StreamSubscription subscription) { - if (_isAddingStream) { - _StreamControllerAddStreamState addState = _varData; - addState.cancel(); - } - _varData = null; - _state = - (_state & ~(_STATE_SUBSCRIBED | _STATE_ADDSTREAM)) | _STATE_CANCELED; + assert(identical(_subscription, subscription)); + _subscription = null; + _state |= _STATE_CANCELLED; _runGuarded(_onCancel); - if (_doneFuture != null && _doneFuture._mayComplete) { - _doneFuture._asyncSetValue(null); - } } void _recordPause(StreamSubscription subscription) { - if (_isAddingStream) { - _StreamControllerAddStreamState addState = _varData; - addState.pause(); - } _runGuarded(_onPause); } void _recordResume(StreamSubscription subscription) { - if (_isAddingStream) { - _StreamControllerAddStreamState addState = _varData; - addState.resume(); - } _runGuarded(_onResume); } } -abstract class _SyncStreamControllerDispatch - implements _StreamController { +class _SyncStreamController extends _StreamController { + _SyncStreamController(void onListen(), + void onPause(), + void onResume(), + void onCancel()) + : super(onListen, onPause, onResume, onCancel); + void _sendData(T data) { _subscription._add(data); } @@ -515,8 +330,13 @@ abstract class _SyncStreamControllerDispatch } } -abstract class _AsyncStreamControllerDispatch - implements _StreamController { +class _AsyncStreamController extends _StreamController { + _AsyncStreamController(void onListen(), + void onPause(), + void onResume(), + void onCancel()) + : super(onListen, onPause, onResume, onCancel); + void _sendData(T data) { _subscription._addPending(new _DelayedData(data)); } @@ -530,48 +350,6 @@ abstract class _AsyncStreamControllerDispatch } } -// TODO(lrn): Use common superclass for callback-controllers when VM supports -// constructors in mixin superclasses. - -class _AsyncStreamController extends _StreamController - with _AsyncStreamControllerDispatch { - final _NotificationHandler _onListen; - final _NotificationHandler _onPause; - final _NotificationHandler _onResume; - final _NotificationHandler _onCancel; - - _AsyncStreamController(void this._onListen(), - void this._onPause(), - void this._onResume(), - void this._onCancel()); -} - -class _SyncStreamController extends _StreamController - with _SyncStreamControllerDispatch { - final _NotificationHandler _onListen; - final _NotificationHandler _onPause; - final _NotificationHandler _onResume; - final _NotificationHandler _onCancel; - - _SyncStreamController(void this._onListen(), - void this._onPause(), - void this._onResume(), - void this._onCancel()); -} - -abstract class _NoCallbacks { - _NotificationHandler get _onListen => null; - _NotificationHandler get _onPause => null; - _NotificationHandler get _onResume => null; - _NotificationHandler get _onCancel => null; -} - -typedef _NoCallbackAsyncStreamController/**/ = _StreamController/**/ - with _AsyncStreamControllerDispatch/**/, _NoCallbacks; - -typedef _NoCallbackSyncStreamController/**/ = _StreamController/**/ - with _SyncStreamControllerDispatch/**/, _NoCallbacks; - typedef void _NotificationHandler(); void _runGuarded(_NotificationHandler notificationHandler) { @@ -585,6 +363,7 @@ void _runGuarded(_NotificationHandler notificationHandler) { class _ControllerStream extends _StreamImpl { _StreamControllerLifecycle _controller; + bool _hasListener = false; _ControllerStream(this._controller); @@ -592,19 +371,17 @@ class _ControllerStream extends _StreamImpl { void onData(T data), void onError(Object error), void onDone(), - bool cancelOnError) => - _controller._subscribe(onData, onError, onDone, cancelOnError); + bool cancelOnError) { + if (_hasListener) { + throw new StateError("The stream has already been listened to."); + } + _hasListener = true; + return new _ControllerSubscription( + _controller, onData, onError, onDone, cancelOnError); + } - // Override == and hashCode so that new streams returned by the same - // controller are considered equal. The controller returns a new stream - // each time it's queried, but doesn't have to cache the result. - - int get hashCode => _controller.hashCode ^ 0x35323532; - - bool operator==(Object other) { - if (other is! _ControllerStream) return false; - _ControllerStream otherStream = other; - return identical(otherStream._controller, this); + void _onListen(_BufferingStreamSubscription subscription) { + _controller._recordListen(subscription); } } @@ -631,64 +408,367 @@ class _ControllerSubscription extends _BufferingStreamSubscription { } } +class _BroadcastStream extends _StreamImpl { + _BroadcastStreamController _controller; -/** A class that exposes only the [StreamSink] interface of an object. */ -class _StreamSinkWrapper implements StreamSink { - final StreamSink _target; - _StreamSinkWrapper(this._target); - void add(T data) { _target.add(data); } - void addError(Object error) { _target.addError(error); } - Future close() => _target.close(); - Future addStream(Stream source) => _target.addStream(source); - Future get done => _target.done; -} + _BroadcastStream(this._controller); -/** - * Object containing the state used to handle [StreamController.addStream]. - */ -class _AddStreamState { - // [_FutureImpl] returned by call to addStream. - _FutureImpl addStreamFuture; + bool get isBroadcast => true; - // Subscription on stream argument to addStream. - StreamSubscription addSubscription; - - _AddStreamState(StreamSink controller, Stream source) - : addStreamFuture = new _FutureImpl(), - addSubscription = source.listen(controller._add, - onError: controller._addError, - onDone: controller._close, - cancelOnError: true); - - void pause() { - addSubscription.pause(); + StreamSubscription _createSubscription( + void onData(T data), + void onError(Object error), + void onDone(), + bool cancelOnError) { + return new _BroadcastSubscription( + _controller, onData, onError, onDone, cancelOnError); } - void resume() { - addSubscription.resume(); - } - - void cancel() { - addSubscription.cancel(); - complete(); - } - - void complete() { - addStreamFuture._asyncSetValue(null); + void _onListen(_BufferingStreamSubscription subscription) { + _controller._recordListen(subscription); } } -class _StreamControllerAddStreamState extends _AddStreamState { - // The subscription or pending data of a _StreamController. - // Stored here because we reuse the `_varData` field in the _StreamController - // to store this state object. - var varData; +abstract class _BroadcastSubscriptionLink { + _BroadcastSubscriptionLink _next; + _BroadcastSubscriptionLink _previous; +} - _StreamControllerAddStreamState(_StreamController controller, - this.varData, - Stream source) : super(controller, source) { - if (controller.isPaused) { - addSubscription.pause(); +class _BroadcastSubscription extends _ControllerSubscription + implements _BroadcastSubscriptionLink { + static const int _STATE_EVENT_ID = 1; + static const int _STATE_FIRING = 2; + static const int _STATE_REMOVE_AFTER_FIRING = 4; + int _eventState; + + _BroadcastSubscriptionLink _next; + _BroadcastSubscriptionLink _previous; + + _BroadcastSubscription(_StreamControllerLifecycle controller, + void onData(T data), + void onError(Object error), + void onDone(), + bool cancelOnError) + : super(controller, onData, onError, onDone, cancelOnError) { + _next = _previous = this; + } + + _BroadcastStreamController get _controller => super._controller; + + bool _expectsEvent(int eventId) { + return (_eventState & _STATE_EVENT_ID) == eventId; + } + + void _toggleEventId() { + _eventState ^= _STATE_EVENT_ID; + } + + bool get _isFiring => (_eventState & _STATE_FIRING) != 0; + + bool _setRemoveAfterFiring() { + assert(_isFiring); + _eventState |= _STATE_REMOVE_AFTER_FIRING; + } + + bool get _removeAfterFiring => + (_eventState & _STATE_REMOVE_AFTER_FIRING) != 0; +} + + +abstract class _BroadcastStreamController + implements StreamController, + _StreamControllerLifecycle, + _BroadcastSubscriptionLink, + _EventDispatch { + static const int _STATE_INITIAL = 0; + static const int _STATE_EVENT_ID = 1; + static const int _STATE_FIRING = 2; + static const int _STATE_CLOSED = 4; + + final _NotificationHandler _onListen; + final _NotificationHandler _onCancel; + + // State of the controller. + int _state; + + // Double-linked list of active listeners. + _BroadcastSubscriptionLink _next; + _BroadcastSubscriptionLink _previous; + + _BroadcastStreamController(this._onListen, this._onCancel) + : _state = _STATE_INITIAL { + _next = _previous = this; + } + + // StreamController interface. + + Stream get stream => new _BroadcastStream(this); + + EventSink get sink => new _EventSinkView(this); + + bool get isClosed => (_state & _STATE_CLOSED) != 0; + + /** + * A broadcast controller is never paused. + * + * Each receiving stream may be paused individually, and they handle their + * own buffering. + */ + bool get isPaused => false; + + /** Whether there are currently a subscriber on the [Stream]. */ + bool get hasListener => !_isEmpty; + + /** Whether an event is being fired (sent to some, but not all, listeners). */ + bool get _isFiring => (_state & _STATE_FIRING) != 0; + + // Linked list helpers + + bool get _isEmpty => identical(_next, this); + + /** Adds subscription to linked list of active listeners. */ + void _addListener(_BroadcastSubscription subscription) { + _BroadcastSubscriptionLink previous = _previous; + previous._next = subscription; + _previous = subscription._previous; + subscription._previous._next = this; + subscription._previous = previous; + subscription._eventState = (_state & _STATE_EVENT_ID); + } + + void _removeListener(_BroadcastSubscription subscription) { + assert(identical(subscription._controller, this)); + assert(!identical(subscription._next, subscription)); + subscription._previous._next = subscription._next; + subscription._next._previous = subscription._previous; + subscription._next = subscription._previous = subscription; + } + + // _StreamControllerLifecycle interface. + + void _recordListen(_BroadcastSubscription subscription) { + _addListener(subscription); + if (identical(_next, _previous)) { + // Only one listener, so it must be the first listener. + _runGuarded(_onListen); + } + } + + void _recordCancel(_BroadcastSubscription subscription) { + if (subscription._isFiring) { + subscription._setRemoveAfterFiring(); + } else { + _removeListener(subscription); + // If we are currently firing an event, the empty-check is performed at + // the end of the listener loop instead of here. + if ((_state & _STATE_FIRING) == 0 && _isEmpty) { + _callOnCancel(); + } + } + } + + void _recordPause(StreamSubscription subscription) {} + void _recordResume(StreamSubscription subscription) {} + + // EventSink interface. + + void add(T data) { + if (isClosed) { + throw new StateError("Cannot add new events after calling close()"); + } + _sendData(data); + } + + void addError(Object error, [Object stackTrace]) { + if (isClosed) { + throw new StateError("Cannot add new events after calling close()"); + } + if (stackTrace != null) _attachStackTrace(error, stackTrace); + _sendError(error); + } + + void close() { + if (isClosed) { + throw new StateError("Cannot add new events after calling close()"); + } + _state |= _STATE_CLOSED; + _sendDone(); + } + + void _forEachListener( + void action(_BufferingStreamSubscription subscription)) { + if (_isFiring) { + throw new StateError( + "Cannot fire new event. Controller is already firing an event"); + } + if (_isEmpty) return; + + // Get event id of this event. + int id = (_state & _STATE_EVENT_ID); + // Start firing (set the _STATE_FIRING bit). We don't do [_onCancel] + // callbacks while firing, and we prevent reentrancy of this function. + // + // Set [_state]'s event id to the next event's id. + // Any listeners added while firing this event will expect the next event, + // not this one, and won't get notified. + _state ^= _STATE_EVENT_ID | _STATE_FIRING; + _BroadcastSubscriptionLink link = _next; + while (!identical(link, this)) { + _BroadcastSubscription subscription = link; + if (subscription._expectsEvent(id)) { + subscription._eventState |= _BroadcastSubscription._STATE_FIRING; + action(subscription); + subscription._toggleEventId(); + link = subscription._next; + if (subscription._removeAfterFiring) { + _removeListener(subscription); + } + subscription._eventState &= ~_BroadcastSubscription._STATE_FIRING; + } else { + link = subscription._next; + } + } + _state &= ~_STATE_FIRING; + + if (_isEmpty) { + _callOnCancel(); + } + } + + void _callOnCancel() { + _runGuarded(_onCancel); + } +} + +class _SyncBroadcastStreamController extends _BroadcastStreamController { + _SyncBroadcastStreamController(void onListen(), void onCancel()) + : super(onListen, onCancel); + + // EventDispatch interface. + + void _sendData(T data) { + if (_isEmpty) return; + _forEachListener((_BufferingStreamSubscription subscription) { + subscription._add(data); + }); + } + + void _sendError(Object error) { + if (_isEmpty) return; + _forEachListener((_BufferingStreamSubscription subscription) { + subscription._addError(error); + }); + } + + void _sendDone() { + if (_isEmpty) return; + _forEachListener((_BroadcastSubscription subscription) { + subscription._close(); + subscription._eventState |= + _BroadcastSubscription._STATE_REMOVE_AFTER_FIRING; + }); + } +} + +class _AsyncBroadcastStreamController extends _BroadcastStreamController { + _AsyncBroadcastStreamController(void onListen(), void onCancel()) + : super(onListen, onCancel); + + // EventDispatch interface. + + void _sendData(T data) { + for (_BroadcastSubscriptionLink link = _next; + !identical(link, this); + link = link._next) { + _BroadcastSubscription subscription = link; + subscription._addPending(new _DelayedData(data)); + } + } + + void _sendError(Object error) { + for (_BroadcastSubscriptionLink link = _next; + !identical(link, this); + link = link._next) { + _BroadcastSubscription subscription = link; + subscription._addPending(new _DelayedError(error)); + } + } + + void _sendDone() { + for (_BroadcastSubscriptionLink link = _next; + !identical(link, this); + link = link._next) { + _BroadcastSubscription subscription = link; + subscription._addPending(const _DelayedDone()); } } } + +/** + * Stream controller that is used by [Stream.asBroadcastStream]. + * + * This stream controller allows incoming events while it is firing + * other events. This is handled by delaying the events until the + * current event is done firing, and then fire the pending events. + * + * This class extends [_SyncBroadcastStreamController]. Events of + * an "asBroadcastStream" stream are always initiated by events + * on another stream, and it is fine to forward them synchronously. + */ +class _AsBroadcastStreamController + extends _SyncBroadcastStreamController + implements _EventDispatch { + _StreamImplEvents _pending; + + _AsBroadcastStreamController(void onListen(), void onCancel()) + : super(onListen, onCancel); + + bool get _hasPending => _pending != null && ! _pending.isEmpty; + + void _addPendingEvent(_DelayedEvent event) { + if (_pending == null) { + _pending = new _StreamImplEvents(); + } + _pending.add(event); + } + + void add(T data) { + if (_isFiring) { + _addPendingEvent(new _DelayedData(data)); + return; + } + super.add(data); + while (_hasPending) { + _pending.handleNext(this); + } + } + + void addError(Object error, [StackTrace stackTrace]) { + if (_isFiring) { + _addPendingEvent(new _DelayedError(error)); + return; + } + super.addError(error, stackTrace); + while (_hasPending) { + _pending.handleNext(this); + } + } + + void close() { + if (_isFiring) { + _addPendingEvent(const _DelayedDone()); + _state |= _STATE_CLOSED; + return; + } + super.close(); + assert(!_hasPending); + } + + void _callOnCancel() { + if (_hasPending) { + _pending.clear(); + _pending = null; + } + super._callOnCancel(); + } +} diff --git a/sdk/lib/async/stream_impl.dart b/sdk/lib/async/stream_impl.dart index 108e42b7d7d..85aa0833b6f 100644 --- a/sdk/lib/async/stream_impl.dart +++ b/sdk/lib/async/stream_impl.dart @@ -719,11 +719,8 @@ class _AsBroadcastStream extends Stream { onError: _controller.addError, onDone: _controller.close); } - if (onData == null) onData = _nullDataHandler; - if (onError == null) onError = _nullErrorHandler; - if (onDone == null) onDone = _nullDoneHandler; - cancelOnError = identical(true, cancelOnError); - return _controller._subscribe(onData, onError, onDone, cancelOnError); + return _controller.stream.listen(onData, onError: onError, onDone: onDone, + cancelOnError: cancelOnError); } void _onCancel() { diff --git a/tests/lib/async/event_helper.dart b/tests/lib/async/event_helper.dart index d0e5b82133a..eeafcc69ff9 100644 --- a/tests/lib/async/event_helper.dart +++ b/tests/lib/async/event_helper.dart @@ -61,10 +61,8 @@ class DoneEvent implements Event { /** Collector of events. */ class Events implements EventSink { final List events = []; - bool trace = false; Events(); - Events.fromIterable(Iterable iterable) { for (var value in iterable) add(value); close(); @@ -76,17 +74,14 @@ class Events implements EventSink { // EventSink interface. void add(var value) { - if (trace) print("Events#$hashCode: add($value)"); events.add(new DataEvent(value)); } void addError(error) { - if (trace) print("Events#$hashCode: addError($error)"); events.add(new ErrorEvent(error)); } void close() { - if (trace) print("Events#$hashCode: close()"); events.add(const DoneEvent()); } @@ -162,17 +157,14 @@ class CaptureEvents extends Events { } void pause([Future resumeSignal]) { - if (trace) print("Events#$hashCode: pause"); subscription.pause(resumeSignal); } void resume() { - if (trace) print("Events#$hashCode: resume"); subscription.resume(); } void onDone(void action()) { - if (trace) print("Events#$hashCode: onDone"); onDoneSignal.future.whenComplete(action); } } diff --git a/tests/lib/async/stream_controller_async_test.dart b/tests/lib/async/stream_controller_async_test.dart index e97ab2c4e4f..7c134055bc2 100644 --- a/tests/lib/async/stream_controller_async_test.dart +++ b/tests/lib/async/stream_controller_async_test.dart @@ -464,6 +464,7 @@ void testBroadcastController() { test("broadcast-controller-individual-pause", () { StreamProtocolTest test = new StreamProtocolTest.broadcast(); + test.trace = true; var sub1; test..expectListen() ..expectData(42) @@ -497,114 +498,6 @@ void testBroadcastController() { }); } -void testSink({bool sync, bool broadcast, bool asBroadcast}) { - String type = "${sync?"S":"A"}${broadcast?"B":"S"}${asBroadcast?"aB":""}"; - test("$type-controller-sink", () { - var done = expectAsync0((){}); - var c = broadcast ? new StreamController.broadcast(sync: sync) - : new StreamController(sync: sync); - var expected = new Events() - ..add(42)..error("error") - ..add(1)..add(2)..add(3)..add(4)..add(5) - ..add(43)..close(); - var actual = new Events.capture(asBroadcast ? c.stream.asBroadcastStream() - : c.stream); - var sink = c.sink; - sink.add(42); - sink.addError("error"); - sink.addStream(new Stream.fromIterable([1, 2, 3, 4, 5])) - .then((_) { - sink.add(43); - return sink.close(); - }) - .then((_) { - Expect.listEquals(expected.events, actual.events); - done(); - }); - }); - - test("$type-controller-sink-canceled", () { - var done = expectAsync0((){}); - var c = broadcast ? new StreamController.broadcast(sync: sync) - : new StreamController(sync: sync); - var expected = new Events() - ..add(42)..error("error") - ..add(1)..add(2)..add(3); - var stream = asBroadcast ? c.stream.asBroadcastStream() : c.stream; - var actual = new Events(); - var sub; - // Cancel subscription after receiving "3" event. - sub = stream.listen((v) { - if (v == 3) sub.cancel(); - actual.add(v); - }, onError: actual.error); - var sink = c.sink; - sink.add(42); - sink.addError("error"); - sink.addStream(new Stream.fromIterable([1, 2, 3, 4, 5])) - .then((_) { - Expect.listEquals(expected.events, actual.events); - // Close controller as well. It has no listener. If it is a broadcast - // stream, it will still be open, and we read the "done" future before - // closing. A normal stream is already done when its listener cancels. - Future doneFuture = sink.done; - sink.close(); - return doneFuture; - }) - .then((_) { - // No change in events. - Expect.listEquals(expected.events, actual.events); - done(); - }); - }); - - test("$type-controller-sink-paused", () { - var done = expectAsync0((){}); - var c = broadcast ? new StreamController.broadcast(sync: sync) - : new StreamController(sync: sync); - var expected = new Events() - ..add(42)..error("error") - ..add(1)..add(2)..add(3) - ..add(4)..add(5)..add(43)..close(); - var stream = asBroadcast ? c.stream.asBroadcastStream() : c.stream; - var actual = new Events(); - var sub; - sub = stream.listen( - (v) { - if (v == 3) { - sub.pause(new Future.delayed(const Duration(milliseconds: 15), - () => null)); - } - actual.add(v); - }, - onError: actual.error, - onDone: actual.close); - var sink = c.sink; - sink.add(42); - sink.addError("error"); - sink.addStream(new Stream.fromIterable([1, 2, 3, 4, 5])) - .then((_) { - sink.add(43); - return sink.close(); - }) - .then((_) { - if (asBroadcast) { - // The done-future of the sink completes when it passes - // the done event to the asBroadcastStream controller, which is - // before the final listener gets the event. - // Wait for the pause to end before testing the events. - return new Future.delayed(const Duration(milliseconds: 50), () { - Expect.listEquals(expected.events, actual.events); - done(); - }); - } else { - Expect.listEquals(expected.events, actual.events); - done(); - } - }); - }); -} - main() { testController(); testSingleController(); @@ -612,10 +505,4 @@ main() { testPause(); testRethrow(); testBroadcastController(); - testSink(sync: true, broadcast: false, asBroadcast: false); - testSink(sync: true, broadcast: false, asBroadcast: true); - testSink(sync: true, broadcast: true, asBroadcast: false); - testSink(sync: false, broadcast: false, asBroadcast: false); - testSink(sync: false, broadcast: false, asBroadcast: true); - testSink(sync: false, broadcast: true, asBroadcast: false); } diff --git a/tests/lib/async/stream_controller_test.dart b/tests/lib/async/stream_controller_test.dart index 439121c10f6..b4aa6aa47ff 100644 --- a/tests/lib/async/stream_controller_test.dart +++ b/tests/lib/async/stream_controller_test.dart @@ -9,7 +9,7 @@ import "package:expect/expect.dart"; import 'dart:async'; import 'event_helper.dart'; -void testMultiController() { +testMultiController() { // Test normal flow. var c = new StreamController(sync: true); Events expectedEvents = new Events() @@ -408,7 +408,7 @@ testExtraMethods() { Expect.listEquals(expectedEvents.events, actualEvents.events); } -void testClosed() { +testClosed() { StreamController c = new StreamController(sync: true); Expect.isFalse(c.isClosed); c.add(42);