dart-lang/sdk
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Revert "Make StreamController be a StreamSink, not just an EventSink."

Browse Source 
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
This commit is contained in:
floitsch@google.com
2013-06-28 15:07:09 +00:00
parent 72219cf60b
commit df2bcb893f
10 changed files with 478 additions and 1026 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sdk/lib/async/async.dart
-1
View File
@@ -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';
sdk/lib/async/async_sources.gypi
-1
View File
@@ -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',
sdk/lib/async/broadcast_stream_controller.dart
-492
View File
@@ -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<T> extends _StreamImpl<T> {
_BroadcastStreamController _controller;
_BroadcastStream(this._controller);
bool get isBroadcast => true;
StreamSubscription<T> _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<T> extends _ControllerSubscription<T>
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<T>
implements StreamController<T>,
_StreamControllerLifecycle<T>,
_BroadcastSubscriptionLink,
_EventSink<T>,
_EventDispatch<T> {
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<T> _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<T> get stream => new _BroadcastStream<T>(this);
StreamSink<T> get sink => new _StreamSinkWrapper<T>(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<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);
}
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;
subscription._next = subscription._previous = subscription;
}
// _StreamControllerLifecycle interface.
StreamSubscription<T> _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<T>(
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<T> 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<T> subscription) {}
void _recordResume(StreamSubscription<T> 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<T> 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<T> 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<T> 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<T> extends _BroadcastStreamController<T> {
_SyncBroadcastStreamController(void onListen(), void onCancel())
: super(onListen, onCancel);
// EventDispatch interface.
void _sendData(T data) {
if (_isEmpty) return;
_forEachListener((_BufferingStreamSubscription<T> subscription) {
subscription._add(data);
});
}
void _sendError(Object error) {
if (_isEmpty) return;
_forEachListener((_BufferingStreamSubscription<T> subscription) {
subscription._addError(error);
});
}
void _sendDone() {
if (!_isEmpty) {
_forEachListener((_BroadcastSubscription<T> subscription) {
subscription._close();
});
} else {
assert(_doneFuture != null);
assert(_doneFuture._mayComplete);
_doneFuture._asyncSetValue(null);
}
}
}
class _AsyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {
_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<T> subscription = link;
subscription._addPending(new _DelayedData(data));
}
}
void _sendError(Object error) {
for (_BroadcastSubscriptionLink link = _next;
!identical(link, this);
link = link._next) {
_BroadcastSubscription<T> subscription = link;
subscription._addPending(new _DelayedError(error));
}
}
void _sendDone() {
if (!_isEmpty) {
for (_BroadcastSubscriptionLink link = _next;
!identical(link, this);
link = link._next) {
_BroadcastSubscription<T> 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<T>
extends _SyncBroadcastStreamController<T>
implements _EventDispatch<T> {
_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<T>(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<T> implements StreamSubscription<T> {
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();
}
sdk/lib/async/future_impl.dart
+12 -35
View File
@@ -49,12 +49,12 @@ abstract class _Completer<T> implements Completer<T> {
class _AsyncCompleter<T> extends _Completer<T> {
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<T> implements _FutureListener<T> {
_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<T> implements Future<T> {
/// [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<T> implements Future<T> {
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<T> implements Future<T> {
}
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<T> implements Future<T> {
}
}
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<T> implements Future<T> {
}
}
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;
sdk/lib/async/stream.dart
+13
View File
@@ -925,6 +925,19 @@ class StreamView<T> extends Stream<T> {
}
}
/**
* [EventSink] wrapper that only exposes the [EventSink] interface.
*/
class _EventSinkView<T> extends EventSink<T> {
final EventSink<T> _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.
sdk/lib/async/stream_controller.dart
+448 -368
View File
@@ -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<T> implements StreamSink<T> {
abstract class StreamController<T> implements EventSink<T> {
/** The stream that this controller is controlling. */
Stream<T> get stream;
@@ -75,17 +75,10 @@ abstract class StreamController<T> implements StreamSink<T> {
void onPause(),
void onResume(),
void onCancel(),
bool sync: false}) {
if (onListen == null && onPause == null &&
onResume == null && onCancel == null) {
return sync
? new _NoCallbackSyncStreamController/*<T>*/()
: new _NoCallbackAsyncStreamController/*<T>*/();
}
return sync
bool sync: false})
=> sync
? new _SyncStreamController<T>(onListen, onPause, onResume, onCancel)
: new _AsyncStreamController<T>(onListen, onPause, onResume, onCancel);
}
/**
* A controller where [stream] can be listened to more than once.
@@ -130,9 +123,9 @@ abstract class StreamController<T> implements StreamSink<T> {
}
/**
* 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<T> get sink;
EventSink<T> get sink;
/**
* Whether the stream is closed for adding more events.
@@ -169,10 +162,7 @@ abstract class StreamController<T> implements StreamSink<T> {
abstract class _StreamControllerLifecycle<T> {
StreamSubscription<T> _subscribe(void onData(T data),
void onError(Object error),
void onDone(),
bool cancelOnError);
void _recordListen(StreamSubscription<T> subscription) {}
void _recordPause(StreamSubscription<T> subscription) {}
void _recordResume(StreamSubscription<T> subscription) {}
void _recordCancel(StreamSubscription<T> subscription) {}
@@ -185,205 +175,81 @@ abstract class _StreamControllerLifecycle<T> {
*/
abstract class _StreamController<T> implements StreamController<T>,
_StreamControllerLifecycle<T>,
_EventSink<T>,
_EventDispatch<T> {
// 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<T> _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<T> _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<T>(this);
}
Stream<T> 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<T> get sink => new _EventSinkView<T>(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<T> get stream => new _ControllerStream(this);
/**
* Returns a view of this object that only exposes the [StreamSink] interface.
*/
StreamSink<T> get sink => new _StreamSinkWrapper<T>(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<T> 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<T>(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<T> implements StreamController<T>,
* 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<T>(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<T> _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<T> subscription) {
assert(_subscription == null);
_subscription = subscription;
subscription._setPendingEvents(_pendingEvents);
_pendingEvents = null;
subscription._guardCallback(() {
_runGuarded(_onListen);
});
return subscription;
}
void _recordCancel(StreamSubscription<T> 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<T> subscription) {
if (_isAddingStream) {
_StreamControllerAddStreamState addState = _varData;
addState.pause();
}
_runGuarded(_onPause);
}
void _recordResume(StreamSubscription<T> subscription) {
if (_isAddingStream) {
_StreamControllerAddStreamState addState = _varData;
addState.resume();
}
_runGuarded(_onResume);
}
}
abstract class _SyncStreamControllerDispatch<T>
implements _StreamController<T> {
class _SyncStreamController<T> extends _StreamController<T> {
_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<T>
}
}
abstract class _AsyncStreamControllerDispatch<T>
implements _StreamController<T> {
class _AsyncStreamController<T> extends _StreamController<T> {
_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<T>
}
}
// TODO(lrn): Use common superclass for callback-controllers when VM supports
// constructors in mixin superclasses.
class _AsyncStreamController<T> extends _StreamController<T>
with _AsyncStreamControllerDispatch<T> {
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<T> extends _StreamController<T>
with _SyncStreamControllerDispatch<T> {
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/*<T>*/ = _StreamController/*<T>*/
with _AsyncStreamControllerDispatch/*<T>*/, _NoCallbacks;
typedef _NoCallbackSyncStreamController/*<T>*/ = _StreamController/*<T>*/
with _SyncStreamControllerDispatch/*<T>*/, _NoCallbacks;
typedef void _NotificationHandler();
void _runGuarded(_NotificationHandler notificationHandler) {
@@ -585,6 +363,7 @@ void _runGuarded(_NotificationHandler notificationHandler) {
class _ControllerStream<T> extends _StreamImpl<T> {
_StreamControllerLifecycle<T> _controller;
bool _hasListener = false;
_ControllerStream(this._controller);
@@ -592,19 +371,17 @@ class _ControllerStream<T> extends _StreamImpl<T> {
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<T>(
_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<T> extends _BufferingStreamSubscription<T> {
}
}
class _BroadcastStream<T> extends _StreamImpl<T> {
_BroadcastStreamController _controller;
/** A class that exposes only the [StreamSink] interface of an object. */
class _StreamSinkWrapper<T> implements StreamSink<T> {
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<T> source) => _target.addStream(source);
Future get done => _target.done;
}
_BroadcastStream(this._controller);
/**
* Object containing the state used to handle [StreamController.addStream].
*/
class _AddStreamState<T> {
// [_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<T> _createSubscription(
void onData(T data),
void onError(Object error),
void onDone(),
bool cancelOnError) {
return new _BroadcastSubscription<T>(
_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<T> extends _AddStreamState<T> {
// 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<T> extends _ControllerSubscription<T>
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<T>
implements StreamController<T>,
_StreamControllerLifecycle<T>,
_BroadcastSubscriptionLink,
_EventDispatch<T> {
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<T> get stream => new _BroadcastStream<T>(this);
EventSink<T> get sink => new _EventSinkView<T>(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<T> 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<T> 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<T> subscription) {
_addListener(subscription);
if (identical(_next, _previous)) {
// Only one listener, so it must be the first listener.
_runGuarded(_onListen);
}
}
void _recordCancel(_BroadcastSubscription<T> 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<T> subscription) {}
void _recordResume(StreamSubscription<T> 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<T> 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<T> 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<T> extends _BroadcastStreamController<T> {
_SyncBroadcastStreamController(void onListen(), void onCancel())
: super(onListen, onCancel);
// EventDispatch interface.
void _sendData(T data) {
if (_isEmpty) return;
_forEachListener((_BufferingStreamSubscription<T> subscription) {
subscription._add(data);
});
}
void _sendError(Object error) {
if (_isEmpty) return;
_forEachListener((_BufferingStreamSubscription<T> subscription) {
subscription._addError(error);
});
}
void _sendDone() {
if (_isEmpty) return;
_forEachListener((_BroadcastSubscription<T> subscription) {
subscription._close();
subscription._eventState |=
_BroadcastSubscription._STATE_REMOVE_AFTER_FIRING;
});
}
}
class _AsyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {
_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<T> subscription = link;
subscription._addPending(new _DelayedData(data));
}
}
void _sendError(Object error) {
for (_BroadcastSubscriptionLink link = _next;
!identical(link, this);
link = link._next) {
_BroadcastSubscription<T> subscription = link;
subscription._addPending(new _DelayedError(error));
}
}
void _sendDone() {
for (_BroadcastSubscriptionLink link = _next;
!identical(link, this);
link = link._next) {
_BroadcastSubscription<T> 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<T>
extends _SyncBroadcastStreamController<T>
implements _EventDispatch<T> {
_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<T>(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();
}
}
sdk/lib/async/stream_impl.dart
+2 -5
View File
@@ -719,11 +719,8 @@ class _AsBroadcastStream<T> extends Stream<T> {
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() {
tests/lib/async/event_helper.dart
-8
View File
@@ -61,10 +61,8 @@ class DoneEvent implements Event {
/** Collector of events. */
class Events implements EventSink {
final List<Event> 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);
}
}
tests/lib/async/stream_controller_async_test.dart
+1 -114
View File
@@ -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);
}
tests/lib/async/stream_controller_test.dart
+2 -2
View File
@@ -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);