| Index: tool/input_sdk/lib/async/broadcast_stream_controller.dart
|
| diff --git a/tool/input_sdk/lib/async/broadcast_stream_controller.dart b/tool/input_sdk/lib/async/broadcast_stream_controller.dart
|
| index 401cf17c80ef978fa2e32c9e519b08de0d8c11f9..c34ec9f78b8c17d9afb4bf11a35f64d30f579405 100644
|
| --- a/tool/input_sdk/lib/async/broadcast_stream_controller.dart
|
| +++ b/tool/input_sdk/lib/async/broadcast_stream_controller.dart
|
| @@ -5,30 +5,25 @@
|
| part of dart.async;
|
|
|
| class _BroadcastStream<T> extends _ControllerStream<T> {
|
| - _BroadcastStream(_StreamControllerLifecycle controller) : super(controller);
|
| + _BroadcastStream(_StreamControllerLifecycle<T> controller)
|
| + : super(controller);
|
|
|
| bool get isBroadcast => true;
|
| }
|
|
|
| -abstract class _BroadcastSubscriptionLink {
|
| - _BroadcastSubscriptionLink _next;
|
| - _BroadcastSubscriptionLink _previous;
|
| -}
|
| -
|
| -class _BroadcastSubscription<T> extends _ControllerSubscription<T>
|
| - implements _BroadcastSubscriptionLink {
|
| +class _BroadcastSubscription<T> extends _ControllerSubscription<T> {
|
| 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;
|
| + int _eventState = 0; // Initialized to help dart2js type inference.
|
|
|
| - _BroadcastSubscriptionLink _next;
|
| - _BroadcastSubscriptionLink _previous;
|
| + _BroadcastSubscription<T> _next;
|
| + _BroadcastSubscription<T> _previous;
|
|
|
| - _BroadcastSubscription(_StreamControllerLifecycle controller,
|
| + _BroadcastSubscription(_StreamControllerLifecycle<T> controller,
|
| void onData(T data),
|
| Function onError,
|
| void onDone(),
|
| @@ -65,11 +60,9 @@ class _BroadcastSubscription<T> extends _ControllerSubscription<T>
|
| // _onCancel is inherited.
|
| }
|
|
|
| -
|
| abstract class _BroadcastStreamController<T>
|
| implements StreamController<T>,
|
| _StreamControllerLifecycle<T>,
|
| - _BroadcastSubscriptionLink,
|
| _EventSink<T>,
|
| _EventDispatch<T> {
|
| static const int _STATE_INITIAL = 0;
|
| @@ -78,15 +71,15 @@ abstract class _BroadcastStreamController<T>
|
| static const int _STATE_CLOSED = 4;
|
| static const int _STATE_ADDSTREAM = 8;
|
|
|
| - final _NotificationHandler _onListen;
|
| - final _NotificationHandler _onCancel;
|
| + ControllerCallback onListen;
|
| + ControllerCancelCallback onCancel;
|
|
|
| // State of the controller.
|
| int _state;
|
|
|
| // Double-linked list of active listeners.
|
| - _BroadcastSubscriptionLink _next;
|
| - _BroadcastSubscriptionLink _previous;
|
| + _BroadcastSubscription<T> _firstSubscription;
|
| + _BroadcastSubscription<T> _lastSubscription;
|
|
|
| // Extra state used during an [addStream] call.
|
| _AddStreamState<T> _addStreamState;
|
| @@ -106,9 +99,27 @@ abstract class _BroadcastStreamController<T>
|
| */
|
| _Future _doneFuture;
|
|
|
| - _BroadcastStreamController(this._onListen, this._onCancel)
|
| - : _state = _STATE_INITIAL {
|
| - _next = _previous = this;
|
| + _BroadcastStreamController(this.onListen, this.onCancel)
|
| + : _state = _STATE_INITIAL;
|
| +
|
| + ControllerCallback get onPause {
|
| + throw new UnsupportedError(
|
| + "Broadcast stream controllers do not support pause callbacks");
|
| + }
|
| +
|
| + void set onPause(void onPauseHandler()) {
|
| + throw new UnsupportedError(
|
| + "Broadcast stream controllers do not support pause callbacks");
|
| + }
|
| +
|
| + ControllerCallback get onResume {
|
| + throw new UnsupportedError(
|
| + "Broadcast stream controllers do not support pause callbacks");
|
| + }
|
| +
|
| + void set onResume(void onResumeHandler()) {
|
| + throw new UnsupportedError(
|
| + "Broadcast stream controllers do not support pause callbacks");
|
| }
|
|
|
| // StreamController interface.
|
| @@ -137,7 +148,7 @@ abstract class _BroadcastStreamController<T>
|
| */
|
| bool get _hasOneListener {
|
| assert(!_isEmpty);
|
| - return identical(_next._next, this);
|
| + return identical(_firstSubscription, _lastSubscription);
|
| }
|
|
|
| /** Whether an event is being fired (sent to some, but not all, listeners). */
|
| @@ -154,26 +165,42 @@ abstract class _BroadcastStreamController<T>
|
|
|
| // Linked list helpers
|
|
|
| - bool get _isEmpty => identical(_next, this);
|
| + bool get _isEmpty => _firstSubscription == null;
|
|
|
| /** Adds subscription to linked list of active listeners. */
|
| void _addListener(_BroadcastSubscription<T> 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);
|
| + // Insert in linked list as last subscription.
|
| + _BroadcastSubscription<T> oldLast = _lastSubscription;
|
| + _lastSubscription = subscription;
|
| + subscription._next = null;
|
| + subscription._previous = oldLast;
|
| + if (oldLast == null) {
|
| + _firstSubscription = subscription;
|
| + } else {
|
| + oldLast._next = subscription;
|
| + }
|
| }
|
|
|
| void _removeListener(_BroadcastSubscription<T> 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;
|
| + _BroadcastSubscription<T> previous = subscription._previous;
|
| + _BroadcastSubscription<T> next = subscription._next;
|
| + if (previous == null) {
|
| + // This was the first subscription.
|
| + _firstSubscription = next;
|
| + } else {
|
| + previous._next = next;
|
| + }
|
| + if (next == null) {
|
| + // This was the last subscription.
|
| + _lastSubscription = previous;
|
| + } else {
|
| + next._previous = previous;
|
| + }
|
| +
|
| subscription._next = subscription._previous = subscription;
|
| }
|
|
|
| @@ -188,26 +215,24 @@ abstract class _BroadcastStreamController<T>
|
| if (onDone == null) onDone = _nullDoneHandler;
|
| return new _DoneStreamSubscription<T>(onDone);
|
| }
|
| - StreamSubscription subscription =
|
| + StreamSubscription<T> subscription =
|
| new _BroadcastSubscription<T>(this, onData, onError, onDone,
|
| cancelOnError);
|
| _addListener(subscription);
|
| - if (identical(_next, _previous)) {
|
| + if (identical(_firstSubscription, _lastSubscription)) {
|
| // Only one listener, so it must be the first listener.
|
| - _runGuarded(_onListen);
|
| + _runGuarded(onListen);
|
| }
|
| return subscription;
|
| }
|
|
|
| Future _recordCancel(StreamSubscription<T> sub) {
|
| - var subscription = sub as _BroadcastSubscription<T>;
|
| + _BroadcastSubscription<T> subscription = sub;
|
| // If already removed by the stream, don't remove it again.
|
| if (identical(subscription._next, subscription)) return null;
|
| - 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.
|
| @@ -296,27 +321,27 @@ abstract class _BroadcastStreamController<T>
|
|
|
| // Get event id of this event.
|
| int id = (_state & _STATE_EVENT_ID);
|
| - // Start firing (set the _STATE_FIRING bit). We don't do [_onCancel]
|
| + // 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<T> subscription = link;
|
| + _BroadcastSubscription<T> subscription = _firstSubscription;
|
| + while (subscription != null) {
|
| if (subscription._expectsEvent(id)) {
|
| subscription._eventState |= _BroadcastSubscription._STATE_FIRING;
|
| action(subscription);
|
| subscription._toggleEventId();
|
| - link = subscription._next;
|
| + _BroadcastSubscription<T> next = subscription._next;
|
| if (subscription._removeAfterFiring) {
|
| _removeListener(subscription);
|
| }
|
| subscription._eventState &= ~_BroadcastSubscription._STATE_FIRING;
|
| + subscription = next;
|
| } else {
|
| - link = subscription._next;
|
| + subscription = subscription._next;
|
| }
|
| }
|
| _state &= ~_STATE_FIRING;
|
| @@ -332,21 +357,32 @@ abstract class _BroadcastStreamController<T>
|
| // When closed, _doneFuture is not null.
|
| _doneFuture._asyncComplete(null);
|
| }
|
| - _runGuarded(_onCancel);
|
| + _runGuarded(onCancel);
|
| }
|
| }
|
|
|
| -class _SyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {
|
| +class _SyncBroadcastStreamController<T> extends _BroadcastStreamController<T>
|
| + implements SynchronousStreamController<T> {
|
| _SyncBroadcastStreamController(void onListen(), void onCancel())
|
| : super(onListen, onCancel);
|
|
|
| // EventDispatch interface.
|
|
|
| + bool get _mayAddEvent => super._mayAddEvent && !_isFiring;
|
| +
|
| + _addEventError() {
|
| + if (_isFiring) {
|
| + return new StateError(
|
| + "Cannot fire new event. Controller is already firing an event");
|
| + }
|
| + return super._addEventError();
|
| + }
|
| +
|
| void _sendData(T data) {
|
| if (_isEmpty) return;
|
| if (_hasOneListener) {
|
| _state |= _BroadcastStreamController._STATE_FIRING;
|
| - _BroadcastSubscription subscription = _next;
|
| + _BroadcastSubscription<T> subscription = _firstSubscription;
|
| subscription._add(data);
|
| _state &= ~_BroadcastStreamController._STATE_FIRING;
|
| if (_isEmpty) {
|
| @@ -368,7 +404,7 @@ class _SyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {
|
|
|
| void _sendDone() {
|
| if (!_isEmpty) {
|
| - _forEachListener((_BroadcastSubscription<T> subscription) {
|
| + _forEachListener((_BufferingStreamSubscription<T> subscription) {
|
| subscription._close();
|
| });
|
| } else {
|
| @@ -386,29 +422,26 @@ class _AsyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {
|
| // EventDispatch interface.
|
|
|
| void _sendData(T data) {
|
| - for (_BroadcastSubscriptionLink link = _next;
|
| - !identical(link, this);
|
| - link = link._next) {
|
| - _BroadcastSubscription<T> subscription = link;
|
| - subscription._addPending(new _DelayedData(data));
|
| + for (_BroadcastSubscription<T> subscription = _firstSubscription;
|
| + subscription != null;
|
| + subscription = subscription._next) {
|
| + subscription._addPending(new _DelayedData<T>(data));
|
| }
|
| }
|
|
|
| void _sendError(Object error, StackTrace stackTrace) {
|
| - for (_BroadcastSubscriptionLink link = _next;
|
| - !identical(link, this);
|
| - link = link._next) {
|
| - _BroadcastSubscription<T> subscription = link;
|
| + for (_BroadcastSubscription<T> subscription = _firstSubscription;
|
| + subscription != null;
|
| + subscription = subscription._next) {
|
| subscription._addPending(new _DelayedError(error, stackTrace));
|
| }
|
| }
|
|
|
| void _sendDone() {
|
| if (!_isEmpty) {
|
| - for (_BroadcastSubscriptionLink link = _next;
|
| - !identical(link, this);
|
| - link = link._next) {
|
| - _BroadcastSubscription<T> subscription = link;
|
| + for (_BroadcastSubscription<T> subscription = _firstSubscription;
|
| + subscription != null;
|
| + subscription = subscription._next) {
|
| subscription._addPending(const _DelayedDone());
|
| }
|
| } else {
|
| @@ -433,7 +466,7 @@ class _AsyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {
|
| class _AsBroadcastStreamController<T>
|
| extends _SyncBroadcastStreamController<T>
|
| implements _EventDispatch<T> {
|
| - _StreamImplEvents _pending;
|
| + _StreamImplEvents<T> _pending;
|
|
|
| _AsBroadcastStreamController(void onListen(), void onCancel())
|
| : super(onListen, onCancel);
|
| @@ -442,7 +475,7 @@ class _AsBroadcastStreamController<T>
|
|
|
| void _addPendingEvent(_DelayedEvent event) {
|
| if (_pending == null) {
|
| - _pending = new _StreamImplEvents();
|
| + _pending = new _StreamImplEvents<T>();
|
| }
|
| _pending.add(event);
|
| }
|
| @@ -507,5 +540,5 @@ class _DoneSubscription<T> implements StreamSubscription<T> {
|
| }
|
| Future cancel() { return new _Future.immediate(null); }
|
| bool get isPaused => _pauseCount > 0;
|
| - Future asFuture([Object value]) => new _Future();
|
| + Future/*<E>*/ asFuture/*<E>*/([Object/*=E*/ value]) => new _Future/*<E>*/();
|
| }
|
|
|