| Index: test/dart_codegen/expect/async/stream_impl.dart
|
| diff --git a/test/dart_codegen/expect/async/stream_impl.dart b/test/dart_codegen/expect/async/stream_impl.dart
|
| index 88b7dfb94d5076bfeb2293a3d4b3ea51e87e5f11..a8590a779540a7d6a8950e45af6ed1513e07d031 100644
|
| --- a/test/dart_codegen/expect/async/stream_impl.dart
|
| +++ b/test/dart_codegen/expect/async/stream_impl.dart
|
| @@ -1,783 +1,678 @@
|
| part of dart.async;
|
| -
|
| -abstract class _EventSink<T> {
|
| - void _add(T data);
|
| - void _addError(Object error, StackTrace stackTrace);
|
| - void _close();
|
| -}
|
| -abstract class _EventDispatch<T> {
|
| - void _sendData(T data);
|
| - void _sendError(Object error, StackTrace stackTrace);
|
| - void _sendDone();
|
| -}
|
| -class _BufferingStreamSubscription<T>
|
| - implements StreamSubscription<T>, _EventSink<T>, _EventDispatch<T> {
|
| - static const int _STATE_CANCEL_ON_ERROR = 1;
|
| - static const int _STATE_CLOSED = 2;
|
| - static const int _STATE_INPUT_PAUSED = 4;
|
| - static const int _STATE_CANCELED = 8;
|
| - static const int _STATE_WAIT_FOR_CANCEL = 16;
|
| - static const int _STATE_IN_CALLBACK = 32;
|
| - static const int _STATE_HAS_PENDING = 64;
|
| - static const int _STATE_PAUSE_COUNT = 128;
|
| - static const int _STATE_PAUSE_COUNT_SHIFT = 7;
|
| - _DataHandler<T> _onData;
|
| - Function _onError;
|
| - _DoneHandler _onDone;
|
| - final Zone _zone = Zone.current;
|
| - int _state;
|
| - Future _cancelFuture;
|
| - _PendingEvents _pending;
|
| - _BufferingStreamSubscription(
|
| - void onData(T data), Function onError, void onDone(), bool cancelOnError)
|
| - : _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) {
|
| - this.onData(onData);
|
| - this.onError(onError);
|
| - this.onDone(onDone);
|
| - }
|
| - void _setPendingEvents(_PendingEvents pendingEvents) {
|
| - assert(_pending == null);
|
| - if (pendingEvents == null) return;
|
| - _pending = pendingEvents;
|
| - if (!pendingEvents.isEmpty) {
|
| - _state |= _STATE_HAS_PENDING;
|
| - _pending.schedule(this);
|
| - }
|
| - }
|
| - _PendingEvents _extractPending() {
|
| - assert(_isCanceled);
|
| - _PendingEvents events = _pending;
|
| - _pending = null;
|
| - return events;
|
| - }
|
| - void onData(void handleData(T event)) {
|
| - if (handleData == null) handleData = _nullDataHandler;
|
| - _onData = _zone.registerUnaryCallback(DDC$RT.wrap((void f(T __u99)) {
|
| - void c(T x0) => f(DDC$RT.cast(x0, dynamic, T, "CastParam",
|
| - """line 154, column 43 of dart:async/stream_impl.dart: """, x0 is T,
|
| - false));
|
| - return f == null ? null : c;
|
| - }, handleData, DDC$RT.type((__t102<T> _) {}), __t100, "Wrap",
|
| - """line 154, column 43 of dart:async/stream_impl.dart: """,
|
| - handleData is __t100));
|
| - }
|
| - void onError(Function handleError) {
|
| - if (handleError == null) handleError = _nullErrorHandler;
|
| - _onError = _registerErrorHandler(handleError, _zone);
|
| - }
|
| - void onDone(void handleDone()) {
|
| - if (handleDone == null) handleDone = _nullDoneHandler;
|
| - _onDone = _zone.registerCallback(handleDone);
|
| - }
|
| - void pause([Future resumeSignal]) {
|
| - if (_isCanceled) return;
|
| - bool wasPaused = _isPaused;
|
| - bool wasInputPaused = _isInputPaused;
|
| - _state = (_state + _STATE_PAUSE_COUNT) | _STATE_INPUT_PAUSED;
|
| - if (resumeSignal != null) resumeSignal.whenComplete(resume);
|
| - if (!wasPaused && _pending != null) _pending.cancelSchedule();
|
| - if (!wasInputPaused && !_inCallback) _guardCallback(_onPause);
|
| - }
|
| - void resume() {
|
| - if (_isCanceled) return;
|
| - if (_isPaused) {
|
| - _decrementPauseCount();
|
| - if (!_isPaused) {
|
| - if (_hasPending && !_pending.isEmpty) {
|
| - _pending.schedule(this);
|
| - } else {
|
| - assert(_mayResumeInput);
|
| - _state &= ~_STATE_INPUT_PAUSED;
|
| - if (!_inCallback) _guardCallback(_onResume);
|
| - }
|
| - }
|
| - }
|
| - }
|
| - Future cancel() {
|
| - _state &= ~_STATE_WAIT_FOR_CANCEL;
|
| - if (_isCanceled) return _cancelFuture;
|
| - _cancel();
|
| - return _cancelFuture;
|
| - }
|
| - Future asFuture([var futureValue]) {
|
| - _Future<T> result = new _Future<T>();
|
| - _onDone = () {
|
| - result._complete(futureValue);
|
| - };
|
| - _onError = (error, stackTrace) {
|
| - cancel();
|
| - result._completeError(error, DDC$RT.cast(stackTrace, dynamic, StackTrace,
|
| - "CastGeneral",
|
| - """line 212, column 36 of dart:async/stream_impl.dart: """,
|
| - stackTrace is StackTrace, true));
|
| - };
|
| - return result;
|
| - }
|
| - bool get _isInputPaused => (_state & _STATE_INPUT_PAUSED) != 0;
|
| - bool get _isClosed => (_state & _STATE_CLOSED) != 0;
|
| - bool get _isCanceled => (_state & _STATE_CANCELED) != 0;
|
| - bool get _waitsForCancel => (_state & _STATE_WAIT_FOR_CANCEL) != 0;
|
| - bool get _inCallback => (_state & _STATE_IN_CALLBACK) != 0;
|
| - bool get _hasPending => (_state & _STATE_HAS_PENDING) != 0;
|
| - bool get _isPaused => _state >= _STATE_PAUSE_COUNT;
|
| - bool get _canFire => _state < _STATE_IN_CALLBACK;
|
| - bool get _mayResumeInput =>
|
| - !_isPaused && (_pending == null || _pending.isEmpty);
|
| - bool get _cancelOnError => (_state & _STATE_CANCEL_ON_ERROR) != 0;
|
| - bool get isPaused => _isPaused;
|
| - void _cancel() {
|
| - _state |= _STATE_CANCELED;
|
| - if (_hasPending) {
|
| - _pending.cancelSchedule();
|
| - }
|
| - if (!_inCallback) _pending = null;
|
| - _cancelFuture = _onCancel();
|
| - }
|
| - void _incrementPauseCount() {
|
| - _state = (_state + _STATE_PAUSE_COUNT) | _STATE_INPUT_PAUSED;
|
| - }
|
| - void _decrementPauseCount() {
|
| - assert(_isPaused);
|
| - _state -= _STATE_PAUSE_COUNT;
|
| - }
|
| - void _add(T data) {
|
| - assert(!_isClosed);
|
| - if (_isCanceled) return;
|
| - if (_canFire) {
|
| - _sendData(data);
|
| - } else {
|
| - _addPending(new _DelayedData(data));
|
| - }
|
| - }
|
| - void _addError(Object error, StackTrace stackTrace) {
|
| - if (_isCanceled) return;
|
| - if (_canFire) {
|
| - _sendError(error, stackTrace);
|
| - } else {
|
| - _addPending(new _DelayedError(error, stackTrace));
|
| - }
|
| - }
|
| - void _close() {
|
| - assert(!_isClosed);
|
| - if (_isCanceled) return;
|
| - _state |= _STATE_CLOSED;
|
| - if (_canFire) {
|
| - _sendDone();
|
| - } else {
|
| - _addPending(const _DelayedDone());
|
| - }
|
| - }
|
| - void _onPause() {
|
| - assert(_isInputPaused);
|
| - }
|
| - void _onResume() {
|
| - assert(!_isInputPaused);
|
| - }
|
| - Future _onCancel() {
|
| - assert(_isCanceled);
|
| - return null;
|
| - }
|
| - void _addPending(_DelayedEvent event) {
|
| - _StreamImplEvents pending = DDC$RT.cast(_pending, _PendingEvents,
|
| - _StreamImplEvents, "CastGeneral",
|
| - """line 322, column 33 of dart:async/stream_impl.dart: """,
|
| - _pending is _StreamImplEvents, true);
|
| - if (_pending == null) pending = _pending = new _StreamImplEvents();
|
| - pending.add(event);
|
| - if (!_hasPending) {
|
| - _state |= _STATE_HAS_PENDING;
|
| - if (!_isPaused) {
|
| - _pending.schedule(this);
|
| - }
|
| - }
|
| - }
|
| - void _sendData(T data) {
|
| - assert(!_isCanceled);
|
| - assert(!_isPaused);
|
| - assert(!_inCallback);
|
| - bool wasInputPaused = _isInputPaused;
|
| - _state |= _STATE_IN_CALLBACK;
|
| - _zone.runUnaryGuarded(DDC$RT.wrap((void f(T __u104)) {
|
| - void c(T x0) => f(DDC$RT.cast(x0, dynamic, T, "CastParam",
|
| - """line 341, column 27 of dart:async/stream_impl.dart: """, x0 is T,
|
| - false));
|
| - return f == null ? null : c;
|
| - }, _onData, DDC$RT.type((__t102<T> _) {}), __t100, "Wrap",
|
| - """line 341, column 27 of dart:async/stream_impl.dart: """,
|
| - _onData is __t100), data);
|
| - _state &= ~_STATE_IN_CALLBACK;
|
| - _checkState(wasInputPaused);
|
| - }
|
| - void _sendError(var error, StackTrace stackTrace) {
|
| - assert(!_isCanceled);
|
| - assert(!_isPaused);
|
| - assert(!_inCallback);
|
| - bool wasInputPaused = _isInputPaused;
|
| - void sendError() {
|
| - if (_isCanceled && !_waitsForCancel) return;
|
| - _state |= _STATE_IN_CALLBACK;
|
| - if (_onError is ZoneBinaryCallback) {
|
| - _zone.runBinaryGuarded(DDC$RT.cast(_onError, Function, __t105,
|
| - "CastGeneral",
|
| - """line 358, column 32 of dart:async/stream_impl.dart: """,
|
| - _onError is __t105, false), error, stackTrace);
|
| - } else {
|
| - _zone.runUnaryGuarded(DDC$RT.cast(_onError, Function, __t100,
|
| - "CastGeneral",
|
| - """line 360, column 31 of dart:async/stream_impl.dart: """,
|
| - _onError is __t100, false), error);
|
| - }
|
| - _state &= ~_STATE_IN_CALLBACK;
|
| - }
|
| - if (_cancelOnError) {
|
| - _state |= _STATE_WAIT_FOR_CANCEL;
|
| - _cancel();
|
| - if (_cancelFuture is Future) {
|
| - _cancelFuture.whenComplete(sendError);
|
| - } else {
|
| - sendError();
|
| - }
|
| - } else {
|
| - sendError();
|
| - _checkState(wasInputPaused);
|
| - }
|
| - }
|
| - void _sendDone() {
|
| - assert(!_isCanceled);
|
| - assert(!_isPaused);
|
| - assert(!_inCallback);
|
| - void sendDone() {
|
| - if (!_waitsForCancel) return;
|
| - _state |= (_STATE_CANCELED | _STATE_CLOSED | _STATE_IN_CALLBACK);
|
| - _zone.runGuarded(_onDone);
|
| - _state &= ~_STATE_IN_CALLBACK;
|
| - }
|
| - _cancel();
|
| - _state |= _STATE_WAIT_FOR_CANCEL;
|
| - if (_cancelFuture is Future) {
|
| - _cancelFuture.whenComplete(sendDone);
|
| - } else {
|
| - sendDone();
|
| - }
|
| - }
|
| - void _guardCallback(callback) {
|
| - assert(!_inCallback);
|
| - bool wasInputPaused = _isInputPaused;
|
| - _state |= _STATE_IN_CALLBACK;
|
| - callback();
|
| - _state &= ~_STATE_IN_CALLBACK;
|
| - _checkState(wasInputPaused);
|
| - }
|
| - void _checkState(bool wasInputPaused) {
|
| - assert(!_inCallback);
|
| - if (_hasPending && _pending.isEmpty) {
|
| - _state &= ~_STATE_HAS_PENDING;
|
| - if (_isInputPaused && _mayResumeInput) {
|
| - _state &= ~_STATE_INPUT_PAUSED;
|
| - }
|
| - }
|
| - while (true) {
|
| - if (_isCanceled) {
|
| - _pending = null;
|
| - return;
|
| - }
|
| - bool isInputPaused = _isInputPaused;
|
| - if (wasInputPaused == isInputPaused) break;
|
| - _state ^= _STATE_IN_CALLBACK;
|
| - if (isInputPaused) {
|
| - _onPause();
|
| - } else {
|
| - _onResume();
|
| - }
|
| - _state &= ~_STATE_IN_CALLBACK;
|
| - wasInputPaused = isInputPaused;
|
| - }
|
| - if (_hasPending && !_isPaused) {
|
| - _pending.schedule(this);
|
| - }
|
| - }
|
| + abstract class _EventSink<T> {void _add(T data);
|
| + void _addError(Object error, StackTrace stackTrace);
|
| + void _close();
|
| }
|
| -abstract class _StreamImpl<T> extends Stream<T> {
|
| - StreamSubscription<T> listen(void onData(T data),
|
| - {Function onError, void onDone(), bool cancelOnError}) {
|
| - cancelOnError = identical(true, cancelOnError);
|
| - StreamSubscription subscription =
|
| - _createSubscription(onData, onError, onDone, cancelOnError);
|
| - _onListen(subscription);
|
| - return DDC$RT.cast(subscription,
|
| - DDC$RT.type((StreamSubscription<dynamic> _) {}),
|
| - DDC$RT.type((StreamSubscription<T> _) {}), "CastDynamic",
|
| - """line 476, column 12 of dart:async/stream_impl.dart: """,
|
| - subscription is StreamSubscription<T>, false);
|
| - }
|
| - _BufferingStreamSubscription<T> _createSubscription(void onData(T data),
|
| - Function onError, void onDone(), bool cancelOnError) {
|
| - return new _BufferingStreamSubscription<T>(
|
| - onData, onError, onDone, cancelOnError);
|
| - }
|
| - void _onListen(StreamSubscription subscription) {}
|
| -}
|
| -typedef _PendingEvents _EventGenerator();
|
| -class _GeneratedStreamImpl<T> extends _StreamImpl<T> {
|
| - final _EventGenerator _pending;
|
| - bool _isUsed = false;
|
| - _GeneratedStreamImpl(this._pending);
|
| - StreamSubscription _createSubscription(void onData(T data), Function onError,
|
| - void onDone(), bool cancelOnError) {
|
| - if (_isUsed) throw new StateError("Stream has already been listened to.");
|
| - _isUsed = true;
|
| - return new _BufferingStreamSubscription(
|
| - onData, onError, onDone, cancelOnError).._setPendingEvents(_pending());
|
| - }
|
| + abstract class _EventDispatch<T> {void _sendData(T data);
|
| + void _sendError(Object error, StackTrace stackTrace);
|
| + void _sendDone();
|
| }
|
| -class _IterablePendingEvents<T> extends _PendingEvents {
|
| - Iterator<T> _iterator;
|
| - _IterablePendingEvents(Iterable<T> data) : _iterator = data.iterator;
|
| - bool get isEmpty => _iterator == null;
|
| - void handleNext(_EventDispatch dispatch) {
|
| - if (_iterator == null) {
|
| - throw new StateError("No events pending.");
|
| - }
|
| - bool isDone;
|
| - try {
|
| - isDone = !_iterator.moveNext();
|
| - } catch (e, s) {
|
| - _iterator = null;
|
| - dispatch._sendError(e, s);
|
| - return;
|
| - }
|
| - if (!isDone) {
|
| - dispatch._sendData(_iterator.current);
|
| - } else {
|
| - _iterator = null;
|
| - dispatch._sendDone();
|
| - }
|
| - }
|
| - void clear() {
|
| - if (isScheduled) cancelSchedule();
|
| - _iterator = null;
|
| - }
|
| + class _BufferingStreamSubscription<T> implements StreamSubscription<T>, _EventSink<T>, _EventDispatch<T> {static const int _STATE_CANCEL_ON_ERROR = 1;
|
| + static const int _STATE_CLOSED = 2;
|
| + static const int _STATE_INPUT_PAUSED = 4;
|
| + static const int _STATE_CANCELED = 8;
|
| + static const int _STATE_WAIT_FOR_CANCEL = 16;
|
| + static const int _STATE_IN_CALLBACK = 32;
|
| + static const int _STATE_HAS_PENDING = 64;
|
| + static const int _STATE_PAUSE_COUNT = 128;
|
| + static const int _STATE_PAUSE_COUNT_SHIFT = 7;
|
| + _DataHandler<T> _onData;
|
| + Function _onError;
|
| + _DoneHandler _onDone;
|
| + final Zone _zone = Zone.current;
|
| + int _state;
|
| + Future _cancelFuture;
|
| + _PendingEvents _pending;
|
| + _BufferingStreamSubscription(void onData(T data), Function onError, void onDone(), bool cancelOnError) : _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) {
|
| +this.onData(onData);
|
| + this.onError(onError);
|
| + this.onDone(onDone);
|
| }
|
| -typedef void _DataHandler<T>(T value);
|
| -typedef void _DoneHandler();
|
| -void _nullDataHandler(var value) {}
|
| -void _nullErrorHandler(error, [StackTrace stackTrace]) {
|
| - Zone.current.handleUncaughtError(error, stackTrace);
|
| -}
|
| -void _nullDoneHandler() {}
|
| -abstract class _DelayedEvent {
|
| - _DelayedEvent next;
|
| - void perform(_EventDispatch dispatch);
|
| -}
|
| -class _DelayedData<T> extends _DelayedEvent {
|
| - final T value;
|
| - _DelayedData(this.value);
|
| - void perform(_EventDispatch<T> dispatch) {
|
| - dispatch._sendData(value);
|
| - }
|
| + void _setPendingEvents(_PendingEvents pendingEvents) {
|
| +assert (_pending == null); if (pendingEvents == null) return; _pending = pendingEvents;
|
| + if (!pendingEvents.isEmpty) {
|
| +_state |= _STATE_HAS_PENDING;
|
| + _pending.schedule(this);
|
| }
|
| -class _DelayedError extends _DelayedEvent {
|
| - final error;
|
| - final StackTrace stackTrace;
|
| - _DelayedError(this.error, this.stackTrace);
|
| - void perform(_EventDispatch dispatch) {
|
| - dispatch._sendError(error, stackTrace);
|
| - }
|
| }
|
| -class _DelayedDone implements _DelayedEvent {
|
| - const _DelayedDone();
|
| - void perform(_EventDispatch dispatch) {
|
| - dispatch._sendDone();
|
| - }
|
| - _DelayedEvent get next => null;
|
| - void set next(_DelayedEvent _) {
|
| - throw new StateError("No events after a done.");
|
| - }
|
| + _PendingEvents _extractPending() {
|
| +assert (_isCanceled); _PendingEvents events = _pending;
|
| + _pending = null;
|
| + return events;
|
| }
|
| -abstract class _PendingEvents {
|
| - static const int _STATE_UNSCHEDULED = 0;
|
| - static const int _STATE_SCHEDULED = 1;
|
| - static const int _STATE_CANCELED = 3;
|
| - int _state = _STATE_UNSCHEDULED;
|
| - bool get isEmpty;
|
| - bool get isScheduled => _state == _STATE_SCHEDULED;
|
| - bool get _eventScheduled => _state >= _STATE_SCHEDULED;
|
| - void schedule(_EventDispatch dispatch) {
|
| - if (isScheduled) return;
|
| - assert(!isEmpty);
|
| - if (_eventScheduled) {
|
| - assert(_state == _STATE_CANCELED);
|
| - _state = _STATE_SCHEDULED;
|
| - return;
|
| - }
|
| - scheduleMicrotask(() {
|
| - int oldState = _state;
|
| - _state = _STATE_UNSCHEDULED;
|
| - if (oldState == _STATE_CANCELED) return;
|
| - handleNext(dispatch);
|
| - });
|
| - _state = _STATE_SCHEDULED;
|
| - }
|
| - void cancelSchedule() {
|
| - if (isScheduled) _state = _STATE_CANCELED;
|
| - }
|
| - void handleNext(_EventDispatch dispatch);
|
| - void clear();
|
| -}
|
| -class _StreamImplEvents extends _PendingEvents {
|
| - _DelayedEvent firstPendingEvent = null;
|
| - _DelayedEvent lastPendingEvent = null;
|
| - bool get isEmpty => lastPendingEvent == null;
|
| - void add(_DelayedEvent event) {
|
| - if (lastPendingEvent == null) {
|
| - firstPendingEvent = lastPendingEvent = event;
|
| - } else {
|
| - lastPendingEvent = lastPendingEvent.next = event;
|
| - }
|
| - }
|
| - void handleNext(_EventDispatch dispatch) {
|
| - assert(!isScheduled);
|
| - _DelayedEvent event = firstPendingEvent;
|
| - firstPendingEvent = event.next;
|
| - if (firstPendingEvent == null) {
|
| - lastPendingEvent = null;
|
| - }
|
| - event.perform(dispatch);
|
| - }
|
| - void clear() {
|
| - if (isScheduled) cancelSchedule();
|
| - firstPendingEvent = lastPendingEvent = null;
|
| - }
|
| + void onData(void handleData(T event)) {
|
| +if (handleData == null) handleData = _nullDataHandler;
|
| + _onData = _zone.registerUnaryCallback(DDC$RT.wrap((void f(T __u99)) {
|
| +void c(T x0) => f(DDC$RT.cast(x0, dynamic, T, "CastParam", """line 154, column 43 of dart:async/stream_impl.dart: """, x0 is T, false));
|
| + return f == null ? null : c;
|
| }
|
| -class _BroadcastLinkedList {
|
| - _BroadcastLinkedList _next;
|
| - _BroadcastLinkedList _previous;
|
| - void _unlink() {
|
| - _previous._next = _next;
|
| - _next._previous = _previous;
|
| - _next = _previous = this;
|
| - }
|
| - void _insertBefore(_BroadcastLinkedList newNext) {
|
| - _BroadcastLinkedList newPrevious = newNext._previous;
|
| - newPrevious._next = this;
|
| - newNext._previous = _previous;
|
| - _previous._next = newNext;
|
| - _previous = newPrevious;
|
| - }
|
| +, handleData, DDC$RT.type((__t102<T> _) {
|
| }
|
| -typedef void _broadcastCallback(StreamSubscription subscription);
|
| -class _DoneStreamSubscription<T> implements StreamSubscription<T> {
|
| - static const int _DONE_SENT = 1;
|
| - static const int _SCHEDULED = 2;
|
| - static const int _PAUSED = 4;
|
| - final Zone _zone;
|
| - int _state = 0;
|
| - _DoneHandler _onDone;
|
| - _DoneStreamSubscription(this._onDone) : _zone = Zone.current {
|
| - _schedule();
|
| - }
|
| - bool get _isSent => (_state & _DONE_SENT) != 0;
|
| - bool get _isScheduled => (_state & _SCHEDULED) != 0;
|
| - bool get isPaused => _state >= _PAUSED;
|
| - void _schedule() {
|
| - if (_isScheduled) return;
|
| - _zone.scheduleMicrotask(_sendDone);
|
| - _state |= _SCHEDULED;
|
| - }
|
| - void onData(void handleData(T data)) {}
|
| - void onError(Function handleError) {}
|
| - void onDone(void handleDone()) {
|
| - _onDone = handleDone;
|
| - }
|
| - void pause([Future resumeSignal]) {
|
| - _state += _PAUSED;
|
| - if (resumeSignal != null) resumeSignal.whenComplete(resume);
|
| - }
|
| - void resume() {
|
| - if (isPaused) {
|
| - _state -= _PAUSED;
|
| - if (!isPaused && !_isSent) {
|
| - _schedule();
|
| - }
|
| - }
|
| - }
|
| - Future cancel() => null;
|
| - Future asFuture([futureValue]) {
|
| - _Future result = new _Future();
|
| - _onDone = () {
|
| - result._completeWithValue(null);
|
| - };
|
| - return result;
|
| - }
|
| - void _sendDone() {
|
| - _state &= ~_SCHEDULED;
|
| - if (isPaused) return;
|
| - _state |= _DONE_SENT;
|
| - if (_onDone != null) _zone.runGuarded(_onDone);
|
| - }
|
| +), __t100, "Wrap", """line 154, column 43 of dart:async/stream_impl.dart: """, handleData is __t100));
|
| }
|
| -class _AsBroadcastStream<T> extends Stream<T> {
|
| - final Stream<T> _source;
|
| - final _broadcastCallback _onListenHandler;
|
| - final _broadcastCallback _onCancelHandler;
|
| - final Zone _zone;
|
| - _AsBroadcastStreamController<T> _controller;
|
| - StreamSubscription<T> _subscription;
|
| - _AsBroadcastStream(this._source,
|
| - void onListenHandler(StreamSubscription subscription),
|
| - void onCancelHandler(StreamSubscription subscription))
|
| - : _onListenHandler = Zone.current.registerUnaryCallback(DDC$RT.wrap(
|
| - (void f(StreamSubscription<dynamic> __u108)) {
|
| - void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic,
|
| - DDC$RT.type((StreamSubscription<dynamic> _) {}), "CastParam",
|
| - """line 813, column 63 of dart:async/stream_impl.dart: """,
|
| - x0 is StreamSubscription<dynamic>, true));
|
| - return f == null ? null : c;
|
| - }, onListenHandler, __t109, __t100, "Wrap",
|
| - """line 813, column 63 of dart:async/stream_impl.dart: """,
|
| - onListenHandler is __t100)),
|
| - _onCancelHandler = Zone.current.registerUnaryCallback(DDC$RT.wrap(
|
| - (void f(StreamSubscription<dynamic> __u111)) {
|
| - void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic,
|
| - DDC$RT.type((StreamSubscription<dynamic> _) {}), "CastParam",
|
| - """line 814, column 63 of dart:async/stream_impl.dart: """,
|
| - x0 is StreamSubscription<dynamic>, true));
|
| - return f == null ? null : c;
|
| - }, onCancelHandler, __t109, __t100, "Wrap",
|
| - """line 814, column 63 of dart:async/stream_impl.dart: """,
|
| - onCancelHandler is __t100)),
|
| - _zone = Zone.current {
|
| - _controller = new _AsBroadcastStreamController<T>(_onListen, _onCancel);
|
| - }
|
| - bool get isBroadcast => true;
|
| - StreamSubscription<T> listen(void onData(T data),
|
| - {Function onError, void onDone(), bool cancelOnError}) {
|
| - if (_controller == null || _controller.isClosed) {
|
| - return new _DoneStreamSubscription<T>(onDone);
|
| - }
|
| - if (_subscription == null) {
|
| - _subscription = _source.listen(_controller.add,
|
| - onError: _controller.addError, onDone: _controller.close);
|
| - }
|
| - cancelOnError = identical(true, cancelOnError);
|
| - return _controller._subscribe(onData, onError, onDone, cancelOnError);
|
| - }
|
| - void _onCancel() {
|
| - bool shutdown = (_controller == null) || _controller.isClosed;
|
| - if (_onCancelHandler != null) {
|
| - _zone.runUnary(DDC$RT.wrap((void f(StreamSubscription<dynamic> __u112)) {
|
| - void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic,
|
| - DDC$RT.type((StreamSubscription<dynamic> _) {}), "CastParam",
|
| - """line 842, column 22 of dart:async/stream_impl.dart: """,
|
| - x0 is StreamSubscription<dynamic>, true));
|
| - return f == null ? null : c;
|
| - }, _onCancelHandler, __t109, __t100, "Wrap",
|
| - """line 842, column 22 of dart:async/stream_impl.dart: """,
|
| - _onCancelHandler is __t100), new _BroadcastSubscriptionWrapper(this));
|
| - }
|
| - if (shutdown) {
|
| - if (_subscription != null) {
|
| - _subscription.cancel();
|
| - _subscription = null;
|
| - }
|
| + void onError(Function handleError) {
|
| +if (handleError == null) handleError = _nullErrorHandler;
|
| + _onError = _registerErrorHandler(handleError, _zone);
|
| +}
|
| + void onDone(void handleDone()) {
|
| +if (handleDone == null) handleDone = _nullDoneHandler;
|
| + _onDone = _zone.registerCallback(handleDone);
|
| +}
|
| + void pause([Future resumeSignal]) {
|
| +if (_isCanceled) return; bool wasPaused = _isPaused;
|
| + bool wasInputPaused = _isInputPaused;
|
| + _state = (_state + _STATE_PAUSE_COUNT) | _STATE_INPUT_PAUSED;
|
| + if (resumeSignal != null) resumeSignal.whenComplete(resume);
|
| + if (!wasPaused && _pending != null) _pending.cancelSchedule();
|
| + if (!wasInputPaused && !_inCallback) _guardCallback(_onPause);
|
| +}
|
| + void resume() {
|
| +if (_isCanceled) return; if (_isPaused) {
|
| +_decrementPauseCount();
|
| + if (!_isPaused) {
|
| + if (_hasPending && !_pending.isEmpty) {
|
| + _pending.schedule(this);
|
| }
|
| - }
|
| - void _onListen() {
|
| - if (_onListenHandler != null) {
|
| - _zone.runUnary(DDC$RT.wrap((void f(StreamSubscription<dynamic> __u113)) {
|
| - void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic,
|
| - DDC$RT.type((StreamSubscription<dynamic> _) {}), "CastParam",
|
| - """line 854, column 22 of dart:async/stream_impl.dart: """,
|
| - x0 is StreamSubscription<dynamic>, true));
|
| - return f == null ? null : c;
|
| - }, _onListenHandler, __t109, __t100, "Wrap",
|
| - """line 854, column 22 of dart:async/stream_impl.dart: """,
|
| - _onListenHandler is __t100), new _BroadcastSubscriptionWrapper(this));
|
| + else {
|
| + assert (_mayResumeInput); _state &= ~_STATE_INPUT_PAUSED;
|
| + if (!_inCallback) _guardCallback(_onResume);
|
| }
|
| }
|
| - void _cancelSubscription() {
|
| - if (_subscription == null) return;
|
| - StreamSubscription subscription = _subscription;
|
| - _subscription = null;
|
| - _controller = null;
|
| - subscription.cancel();
|
| - }
|
| - void _pauseSubscription(Future resumeSignal) {
|
| - if (_subscription == null) return;
|
| - _subscription.pause(resumeSignal);
|
| - }
|
| - void _resumeSubscription() {
|
| - if (_subscription == null) return;
|
| - _subscription.resume();
|
| - }
|
| - bool get _isSubscriptionPaused {
|
| - if (_subscription == null) return false;
|
| - return _subscription.isPaused;
|
| - }
|
| }
|
| -class _BroadcastSubscriptionWrapper<T> implements StreamSubscription<T> {
|
| - final _AsBroadcastStream _stream;
|
| - _BroadcastSubscriptionWrapper(this._stream);
|
| - void onData(void handleData(T data)) {
|
| - throw new UnsupportedError(
|
| - "Cannot change handlers of asBroadcastStream source subscription.");
|
| - }
|
| - void onError(void handleError(Object data)) {
|
| - throw new UnsupportedError(
|
| - "Cannot change handlers of asBroadcastStream source subscription.");
|
| - }
|
| - void onDone(void handleDone()) {
|
| - throw new UnsupportedError(
|
| - "Cannot change handlers of asBroadcastStream source subscription.");
|
| - }
|
| - void pause([Future resumeSignal]) {
|
| - _stream._pauseSubscription(resumeSignal);
|
| - }
|
| - void resume() {
|
| - _stream._resumeSubscription();
|
| - }
|
| - Future cancel() {
|
| - _stream._cancelSubscription();
|
| - return null;
|
| - }
|
| - bool get isPaused {
|
| - return _stream._isSubscriptionPaused;
|
| - }
|
| - Future asFuture([var futureValue]) {
|
| - throw new UnsupportedError(
|
| - "Cannot change handlers of asBroadcastStream source subscription.");
|
| - }
|
| }
|
| -class _StreamIteratorImpl<T> implements StreamIterator<T> {
|
| - static const int _STATE_FOUND = 0;
|
| - static const int _STATE_DONE = 1;
|
| - static const int _STATE_MOVING = 2;
|
| - static const int _STATE_EXTRA_DATA = 3;
|
| - static const int _STATE_EXTRA_ERROR = 4;
|
| - static const int _STATE_EXTRA_DONE = 5;
|
| - StreamSubscription _subscription;
|
| - T _current = ((__x114) => DDC$RT.cast(__x114, Null, T, "CastLiteral",
|
| - """line 968, column 16 of dart:async/stream_impl.dart: """, __x114 is T,
|
| - false))(null);
|
| - var _futureOrPrefetch = null;
|
| - int _state = _STATE_FOUND;
|
| - _StreamIteratorImpl(final Stream<T> stream) {
|
| - _subscription = stream.listen(_onData,
|
| - onError: _onError, onDone: _onDone, cancelOnError: true);
|
| + Future cancel() {
|
| +_state &= ~_STATE_WAIT_FOR_CANCEL;
|
| + if (_isCanceled) return _cancelFuture;
|
| + _cancel();
|
| + return _cancelFuture;
|
| +}
|
| + Future asFuture([var futureValue]) {
|
| +_Future<T> result = new _Future<T>();
|
| + _onDone = () {
|
| +result._complete(futureValue);
|
| +}
|
| +;
|
| + _onError = (error, stackTrace) {
|
| +cancel();
|
| + result._completeError(error, DDC$RT.cast(stackTrace, dynamic, StackTrace, "CastGeneral", """line 212, column 36 of dart:async/stream_impl.dart: """, stackTrace is StackTrace, true));
|
| +}
|
| +;
|
| + return result;
|
| +}
|
| + bool get _isInputPaused => (_state & _STATE_INPUT_PAUSED) != 0;
|
| + bool get _isClosed => (_state & _STATE_CLOSED) != 0;
|
| + bool get _isCanceled => (_state & _STATE_CANCELED) != 0;
|
| + bool get _waitsForCancel => (_state & _STATE_WAIT_FOR_CANCEL) != 0;
|
| + bool get _inCallback => (_state & _STATE_IN_CALLBACK) != 0;
|
| + bool get _hasPending => (_state & _STATE_HAS_PENDING) != 0;
|
| + bool get _isPaused => _state >= _STATE_PAUSE_COUNT;
|
| + bool get _canFire => _state < _STATE_IN_CALLBACK;
|
| + bool get _mayResumeInput => !_isPaused && (_pending == null || _pending.isEmpty);
|
| + bool get _cancelOnError => (_state & _STATE_CANCEL_ON_ERROR) != 0;
|
| + bool get isPaused => _isPaused;
|
| + void _cancel() {
|
| +_state |= _STATE_CANCELED;
|
| + if (_hasPending) {
|
| +_pending.cancelSchedule();
|
| +}
|
| + if (!_inCallback) _pending = null;
|
| + _cancelFuture = _onCancel();
|
| +}
|
| + void _incrementPauseCount() {
|
| +_state = (_state + _STATE_PAUSE_COUNT) | _STATE_INPUT_PAUSED;
|
| +}
|
| + void _decrementPauseCount() {
|
| +assert (_isPaused); _state -= _STATE_PAUSE_COUNT;
|
| +}
|
| + void _add(T data) {
|
| +assert (!_isClosed); if (_isCanceled) return; if (_canFire) {
|
| +_sendData(data);
|
| +}
|
| + else {
|
| +_addPending(new _DelayedData(data));
|
| +}
|
| +}
|
| + void _addError(Object error, StackTrace stackTrace) {
|
| +if (_isCanceled) return; if (_canFire) {
|
| +_sendError(error, stackTrace);
|
| +}
|
| + else {
|
| +_addPending(new _DelayedError(error, stackTrace));
|
| +}
|
| +}
|
| + void _close() {
|
| +assert (!_isClosed); if (_isCanceled) return; _state |= _STATE_CLOSED;
|
| + if (_canFire) {
|
| +_sendDone();
|
| +}
|
| + else {
|
| +_addPending(const _DelayedDone());
|
| +}
|
| +}
|
| + void _onPause() {
|
| +assert (_isInputPaused);}
|
| + void _onResume() {
|
| +assert (!_isInputPaused);}
|
| + Future _onCancel() {
|
| +assert (_isCanceled); return null;
|
| +}
|
| + void _addPending(_DelayedEvent event) {
|
| +_StreamImplEvents pending = DDC$RT.cast(_pending, _PendingEvents, _StreamImplEvents, "CastGeneral", """line 322, column 33 of dart:async/stream_impl.dart: """, _pending is _StreamImplEvents, true);
|
| + if (_pending == null) pending = _pending = new _StreamImplEvents();
|
| + pending.add(event);
|
| + if (!_hasPending) {
|
| +_state |= _STATE_HAS_PENDING;
|
| + if (!_isPaused) {
|
| + _pending.schedule(this);
|
| }
|
| - T get current => _current;
|
| - Future<bool> moveNext() {
|
| - if (_state == _STATE_DONE) {
|
| - return new _Future<bool>.immediate(false);
|
| - }
|
| - if (_state == _STATE_MOVING) {
|
| - throw new StateError("Already waiting for next.");
|
| - }
|
| - if (_state == _STATE_FOUND) {
|
| - _state = _STATE_MOVING;
|
| - _current = ((__x115) => DDC$RT.cast(__x115, Null, T, "CastLiteral",
|
| - """line 998, column 18 of dart:async/stream_impl.dart: """,
|
| - __x115 is T, false))(null);
|
| - _futureOrPrefetch = new _Future<bool>();
|
| - return DDC$RT.cast(_futureOrPrefetch, dynamic,
|
| - DDC$RT.type((Future<bool> _) {}), "CastGeneral",
|
| - """line 1000, column 14 of dart:async/stream_impl.dart: """,
|
| - _futureOrPrefetch is Future<bool>, false);
|
| - } else {
|
| - assert(_state >= _STATE_EXTRA_DATA);
|
| - switch (_state) {
|
| - case _STATE_EXTRA_DATA:
|
| - _state = _STATE_FOUND;
|
| - _current = DDC$RT.cast(_futureOrPrefetch, dynamic, T, "CastGeneral",
|
| - """line 1006, column 22 of dart:async/stream_impl.dart: """,
|
| - _futureOrPrefetch is T, false);
|
| - _futureOrPrefetch = null;
|
| - _subscription.resume();
|
| - return new _Future<bool>.immediate(true);
|
| - case _STATE_EXTRA_ERROR:
|
| - AsyncError prefetch = DDC$RT.cast(_futureOrPrefetch, dynamic,
|
| - AsyncError, "CastGeneral",
|
| - """line 1011, column 33 of dart:async/stream_impl.dart: """,
|
| - _futureOrPrefetch is AsyncError, true);
|
| - _clear();
|
| - return new _Future<bool>.immediateError(
|
| - prefetch.error, prefetch.stackTrace);
|
| - case _STATE_EXTRA_DONE:
|
| - _clear();
|
| - return new _Future<bool>.immediate(false);
|
| - }
|
| - }
|
| +}
|
| +}
|
| + void _sendData(T data) {
|
| +assert (!_isCanceled); assert (!_isPaused); assert (!_inCallback); bool wasInputPaused = _isInputPaused;
|
| + _state |= _STATE_IN_CALLBACK;
|
| + _zone.runUnaryGuarded(DDC$RT.wrap((void f(T __u104)) {
|
| +void c(T x0) => f(DDC$RT.cast(x0, dynamic, T, "CastParam", """line 341, column 27 of dart:async/stream_impl.dart: """, x0 is T, false));
|
| + return f == null ? null : c;
|
| +}
|
| +, _onData, DDC$RT.type((__t102<T> _) {
|
| +}
|
| +), __t100, "Wrap", """line 341, column 27 of dart:async/stream_impl.dart: """, _onData is __t100), data);
|
| + _state &= ~_STATE_IN_CALLBACK;
|
| + _checkState(wasInputPaused);
|
| +}
|
| + void _sendError(var error, StackTrace stackTrace) {
|
| +assert (!_isCanceled); assert (!_isPaused); assert (!_inCallback); bool wasInputPaused = _isInputPaused;
|
| + void sendError() {
|
| +if (_isCanceled && !_waitsForCancel) return; _state |= _STATE_IN_CALLBACK;
|
| + if (_onError is ZoneBinaryCallback) {
|
| + _zone.runBinaryGuarded(DDC$RT.cast(_onError, Function, __t105, "CastGeneral", """line 358, column 32 of dart:async/stream_impl.dart: """, _onError is __t105, false), error, stackTrace);
|
| }
|
| - void _clear() {
|
| - _subscription = null;
|
| - _futureOrPrefetch = null;
|
| - _current = ((__x116) => DDC$RT.cast(__x116, Null, T, "CastLiteral",
|
| - """line 1026, column 16 of dart:async/stream_impl.dart: """,
|
| - __x116 is T, false))(null);
|
| - _state = _STATE_DONE;
|
| + else {
|
| + _zone.runUnaryGuarded(DDC$RT.cast(_onError, Function, __t100, "CastGeneral", """line 360, column 31 of dart:async/stream_impl.dart: """, _onError is __t100, false), error);
|
| }
|
| - Future cancel() {
|
| - StreamSubscription subscription = _subscription;
|
| - if (_state == _STATE_MOVING) {
|
| - _Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic,
|
| - DDC$RT.type((_Future<bool> _) {}), "CastGeneral",
|
| - """line 1033, column 31 of dart:async/stream_impl.dart: """,
|
| - _futureOrPrefetch is _Future<bool>, false);
|
| - _clear();
|
| - hasNext._complete(false);
|
| - } else {
|
| - _clear();
|
| - }
|
| - return subscription.cancel();
|
| + _state &= ~_STATE_IN_CALLBACK;
|
| +}
|
| + if (_cancelOnError) {
|
| +_state |= _STATE_WAIT_FOR_CANCEL;
|
| + _cancel();
|
| + if (_cancelFuture is Future) {
|
| + _cancelFuture.whenComplete(sendError);
|
| }
|
| - void _onData(T data) {
|
| - if (_state == _STATE_MOVING) {
|
| - _current = data;
|
| - _Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic,
|
| - DDC$RT.type((_Future<bool> _) {}), "CastGeneral",
|
| - """line 1045, column 31 of dart:async/stream_impl.dart: """,
|
| - _futureOrPrefetch is _Future<bool>, false);
|
| - _futureOrPrefetch = null;
|
| - _state = _STATE_FOUND;
|
| - hasNext._complete(true);
|
| - return;
|
| - }
|
| - _subscription.pause();
|
| - assert(_futureOrPrefetch == null);
|
| - _futureOrPrefetch = data;
|
| - _state = _STATE_EXTRA_DATA;
|
| + else {
|
| + sendError();
|
| }
|
| - void _onError(Object error, [StackTrace stackTrace]) {
|
| - if (_state == _STATE_MOVING) {
|
| - _Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic,
|
| - DDC$RT.type((_Future<bool> _) {}), "CastGeneral",
|
| - """line 1059, column 31 of dart:async/stream_impl.dart: """,
|
| - _futureOrPrefetch is _Future<bool>, false);
|
| - _clear();
|
| - hasNext._completeError(error, stackTrace);
|
| - return;
|
| - }
|
| - _subscription.pause();
|
| - assert(_futureOrPrefetch == null);
|
| - _futureOrPrefetch = new AsyncError(error, stackTrace);
|
| - _state = _STATE_EXTRA_ERROR;
|
| - }
|
| - void _onDone() {
|
| - if (_state == _STATE_MOVING) {
|
| - _Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic,
|
| - DDC$RT.type((_Future<bool> _) {}), "CastGeneral",
|
| - """line 1073, column 31 of dart:async/stream_impl.dart: """,
|
| - _futureOrPrefetch is _Future<bool>, false);
|
| - _clear();
|
| - hasNext._complete(false);
|
| - return;
|
| - }
|
| - _subscription.pause();
|
| - _futureOrPrefetch = null;
|
| - _state = _STATE_EXTRA_DONE;
|
| +}
|
| + else {
|
| +sendError();
|
| + _checkState(wasInputPaused);
|
| +}
|
| +}
|
| + void _sendDone() {
|
| +assert (!_isCanceled); assert (!_isPaused); assert (!_inCallback); void sendDone() {
|
| +if (!_waitsForCancel) return; _state |= (_STATE_CANCELED | _STATE_CLOSED | _STATE_IN_CALLBACK);
|
| + _zone.runGuarded(_onDone);
|
| + _state &= ~_STATE_IN_CALLBACK;
|
| +}
|
| + _cancel();
|
| + _state |= _STATE_WAIT_FOR_CANCEL;
|
| + if (_cancelFuture is Future) {
|
| +_cancelFuture.whenComplete(sendDone);
|
| +}
|
| + else {
|
| +sendDone();
|
| +}
|
| +}
|
| + void _guardCallback(callback) {
|
| +assert (!_inCallback); bool wasInputPaused = _isInputPaused;
|
| + _state |= _STATE_IN_CALLBACK;
|
| + callback();
|
| + _state &= ~_STATE_IN_CALLBACK;
|
| + _checkState(wasInputPaused);
|
| +}
|
| + void _checkState(bool wasInputPaused) {
|
| +assert (!_inCallback); if (_hasPending && _pending.isEmpty) {
|
| +_state &= ~_STATE_HAS_PENDING;
|
| + if (_isInputPaused && _mayResumeInput) {
|
| + _state &= ~_STATE_INPUT_PAUSED;
|
| }
|
| }
|
| -typedef dynamic __t100(dynamic __u101);
|
| -typedef void __t102<T>(T __u103);
|
| -typedef dynamic __t105(dynamic __u106, dynamic __u107);
|
| -typedef void __t109(StreamSubscription<dynamic> __u110);
|
| + while (true) {
|
| +if (_isCanceled) {
|
| + _pending = null;
|
| + return;}
|
| + bool isInputPaused = _isInputPaused;
|
| + if (wasInputPaused == isInputPaused) break;
|
| + _state ^= _STATE_IN_CALLBACK;
|
| + if (isInputPaused) {
|
| + _onPause();
|
| + }
|
| + else {
|
| + _onResume();
|
| + }
|
| + _state &= ~_STATE_IN_CALLBACK;
|
| + wasInputPaused = isInputPaused;
|
| +}
|
| + if (_hasPending && !_isPaused) {
|
| +_pending.schedule(this);
|
| +}
|
| +}
|
| +}
|
| + abstract class _StreamImpl<T> extends Stream<T> {StreamSubscription<T> listen(void onData(T data), {
|
| +Function onError, void onDone(), bool cancelOnError}
|
| +) {
|
| +cancelOnError = identical(true, cancelOnError);
|
| + StreamSubscription subscription = _createSubscription(onData, onError, onDone, cancelOnError);
|
| + _onListen(subscription);
|
| + return DDC$RT.cast(subscription, DDC$RT.type((StreamSubscription<dynamic> _) {
|
| +}
|
| +), DDC$RT.type((StreamSubscription<T> _) {
|
| +}
|
| +), "CastDynamic", """line 476, column 12 of dart:async/stream_impl.dart: """, subscription is StreamSubscription<T>, false);
|
| +}
|
| + _BufferingStreamSubscription<T> _createSubscription(void onData(T data), Function onError, void onDone(), bool cancelOnError) {
|
| +return new _BufferingStreamSubscription<T>(onData, onError, onDone, cancelOnError);
|
| +}
|
| + void _onListen(StreamSubscription subscription) {
|
| +}
|
| +}
|
| + typedef _PendingEvents _EventGenerator();
|
| + class _GeneratedStreamImpl<T> extends _StreamImpl<T> {final _EventGenerator _pending;
|
| + bool _isUsed = false;
|
| + _GeneratedStreamImpl(this._pending);
|
| + StreamSubscription _createSubscription(void onData(T data), Function onError, void onDone(), bool cancelOnError) {
|
| +if (_isUsed) throw new StateError("Stream has already been listened to.");
|
| + _isUsed = true;
|
| + return new _BufferingStreamSubscription(onData, onError, onDone, cancelOnError).._setPendingEvents(_pending());
|
| +}
|
| +}
|
| + class _IterablePendingEvents<T> extends _PendingEvents {Iterator<T> _iterator;
|
| + _IterablePendingEvents(Iterable<T> data) : _iterator = data.iterator;
|
| + bool get isEmpty => _iterator == null;
|
| + void handleNext(_EventDispatch dispatch) {
|
| +if (_iterator == null) {
|
| +throw new StateError("No events pending.");
|
| +}
|
| + bool isDone;
|
| + try {
|
| +isDone = !_iterator.moveNext();
|
| +}
|
| + catch (e, s) {
|
| +_iterator = null;
|
| + dispatch._sendError(e, s);
|
| + return;}
|
| + if (!isDone) {
|
| +dispatch._sendData(_iterator.current);
|
| +}
|
| + else {
|
| +_iterator = null;
|
| + dispatch._sendDone();
|
| +}
|
| +}
|
| + void clear() {
|
| +if (isScheduled) cancelSchedule();
|
| + _iterator = null;
|
| +}
|
| +}
|
| + typedef void _DataHandler<T>(T value);
|
| + typedef void _DoneHandler();
|
| + void _nullDataHandler(var value) {
|
| +}
|
| + void _nullErrorHandler(error, [StackTrace stackTrace]) {
|
| +Zone.current.handleUncaughtError(error, stackTrace);
|
| +}
|
| + void _nullDoneHandler() {
|
| +}
|
| + abstract class _DelayedEvent {_DelayedEvent next;
|
| + void perform(_EventDispatch dispatch);
|
| +}
|
| + class _DelayedData<T> extends _DelayedEvent {final T value;
|
| + _DelayedData(this.value);
|
| + void perform(_EventDispatch<T> dispatch) {
|
| +dispatch._sendData(value);
|
| +}
|
| +}
|
| + class _DelayedError extends _DelayedEvent {final error;
|
| + final StackTrace stackTrace;
|
| + _DelayedError(this.error, this.stackTrace);
|
| + void perform(_EventDispatch dispatch) {
|
| +dispatch._sendError(error, stackTrace);
|
| +}
|
| +}
|
| + class _DelayedDone implements _DelayedEvent {const _DelayedDone();
|
| + void perform(_EventDispatch dispatch) {
|
| +dispatch._sendDone();
|
| +}
|
| + _DelayedEvent get next => null;
|
| + void set next(_DelayedEvent _) {
|
| +throw new StateError("No events after a done.");
|
| +}
|
| +}
|
| + abstract class _PendingEvents {static const int _STATE_UNSCHEDULED = 0;
|
| + static const int _STATE_SCHEDULED = 1;
|
| + static const int _STATE_CANCELED = 3;
|
| + int _state = _STATE_UNSCHEDULED;
|
| + bool get isEmpty;
|
| + bool get isScheduled => _state == _STATE_SCHEDULED;
|
| + bool get _eventScheduled => _state >= _STATE_SCHEDULED;
|
| + void schedule(_EventDispatch dispatch) {
|
| +if (isScheduled) return; assert (!isEmpty); if (_eventScheduled) {
|
| +assert (_state == _STATE_CANCELED); _state = _STATE_SCHEDULED;
|
| + return;}
|
| + scheduleMicrotask(() {
|
| +int oldState = _state;
|
| + _state = _STATE_UNSCHEDULED;
|
| + if (oldState == _STATE_CANCELED) return; handleNext(dispatch);
|
| +}
|
| +);
|
| + _state = _STATE_SCHEDULED;
|
| +}
|
| + void cancelSchedule() {
|
| +if (isScheduled) _state = _STATE_CANCELED;
|
| +}
|
| + void handleNext(_EventDispatch dispatch);
|
| + void clear();
|
| +}
|
| + class _StreamImplEvents extends _PendingEvents {_DelayedEvent firstPendingEvent = null;
|
| + _DelayedEvent lastPendingEvent = null;
|
| + bool get isEmpty => lastPendingEvent == null;
|
| + void add(_DelayedEvent event) {
|
| +if (lastPendingEvent == null) {
|
| +firstPendingEvent = lastPendingEvent = event;
|
| +}
|
| + else {
|
| +lastPendingEvent = lastPendingEvent.next = event;
|
| +}
|
| +}
|
| + void handleNext(_EventDispatch dispatch) {
|
| +assert (!isScheduled); _DelayedEvent event = firstPendingEvent;
|
| + firstPendingEvent = event.next;
|
| + if (firstPendingEvent == null) {
|
| +lastPendingEvent = null;
|
| +}
|
| + event.perform(dispatch);
|
| +}
|
| + void clear() {
|
| +if (isScheduled) cancelSchedule();
|
| + firstPendingEvent = lastPendingEvent = null;
|
| +}
|
| +}
|
| + class _BroadcastLinkedList {_BroadcastLinkedList _next;
|
| + _BroadcastLinkedList _previous;
|
| + void _unlink() {
|
| +_previous._next = _next;
|
| + _next._previous = _previous;
|
| + _next = _previous = this;
|
| +}
|
| + void _insertBefore(_BroadcastLinkedList newNext) {
|
| +_BroadcastLinkedList newPrevious = newNext._previous;
|
| + newPrevious._next = this;
|
| + newNext._previous = _previous;
|
| + _previous._next = newNext;
|
| + _previous = newPrevious;
|
| +}
|
| +}
|
| + typedef void _broadcastCallback(StreamSubscription subscription);
|
| + class _DoneStreamSubscription<T> implements StreamSubscription<T> {static const int _DONE_SENT = 1;
|
| + static const int _SCHEDULED = 2;
|
| + static const int _PAUSED = 4;
|
| + final Zone _zone;
|
| + int _state = 0;
|
| + _DoneHandler _onDone;
|
| + _DoneStreamSubscription(this._onDone) : _zone = Zone.current {
|
| +_schedule();
|
| +}
|
| + bool get _isSent => (_state & _DONE_SENT) != 0;
|
| + bool get _isScheduled => (_state & _SCHEDULED) != 0;
|
| + bool get isPaused => _state >= _PAUSED;
|
| + void _schedule() {
|
| +if (_isScheduled) return; _zone.scheduleMicrotask(_sendDone);
|
| + _state |= _SCHEDULED;
|
| +}
|
| + void onData(void handleData(T data)) {
|
| +}
|
| + void onError(Function handleError) {
|
| +}
|
| + void onDone(void handleDone()) {
|
| +_onDone = handleDone;
|
| +}
|
| + void pause([Future resumeSignal]) {
|
| +_state += _PAUSED;
|
| + if (resumeSignal != null) resumeSignal.whenComplete(resume);
|
| +}
|
| + void resume() {
|
| +if (isPaused) {
|
| +_state -= _PAUSED;
|
| + if (!isPaused && !_isSent) {
|
| +_schedule();
|
| +}
|
| +}
|
| +}
|
| + Future cancel() => null;
|
| + Future asFuture([futureValue]) {
|
| +_Future result = new _Future();
|
| + _onDone = () {
|
| +result._completeWithValue(null);
|
| +}
|
| +;
|
| + return result;
|
| +}
|
| + void _sendDone() {
|
| +_state &= ~_SCHEDULED;
|
| + if (isPaused) return; _state |= _DONE_SENT;
|
| + if (_onDone != null) _zone.runGuarded(_onDone);
|
| +}
|
| +}
|
| + class _AsBroadcastStream<T> extends Stream<T> {final Stream<T> _source;
|
| + final _broadcastCallback _onListenHandler;
|
| + final _broadcastCallback _onCancelHandler;
|
| + final Zone _zone;
|
| + _AsBroadcastStreamController<T> _controller;
|
| + StreamSubscription<T> _subscription;
|
| + _AsBroadcastStream(this._source, void onListenHandler(StreamSubscription subscription), void onCancelHandler(StreamSubscription subscription)) : _onListenHandler = Zone.current.registerUnaryCallback(DDC$RT.wrap((void f(StreamSubscription<dynamic> __u108)) {
|
| +void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic, DDC$RT.type((StreamSubscription<dynamic> _) {
|
| +}
|
| +), "CastParam", """line 813, column 63 of dart:async/stream_impl.dart: """, x0 is StreamSubscription<dynamic>, true));
|
| + return f == null ? null : c;
|
| +}
|
| +, onListenHandler, __t109, __t100, "Wrap", """line 813, column 63 of dart:async/stream_impl.dart: """, onListenHandler is __t100)), _onCancelHandler = Zone.current.registerUnaryCallback(DDC$RT.wrap((void f(StreamSubscription<dynamic> __u111)) {
|
| +void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic, DDC$RT.type((StreamSubscription<dynamic> _) {
|
| +}
|
| +), "CastParam", """line 814, column 63 of dart:async/stream_impl.dart: """, x0 is StreamSubscription<dynamic>, true));
|
| + return f == null ? null : c;
|
| +}
|
| +, onCancelHandler, __t109, __t100, "Wrap", """line 814, column 63 of dart:async/stream_impl.dart: """, onCancelHandler is __t100)), _zone = Zone.current {
|
| +_controller = new _AsBroadcastStreamController<T>(_onListen, _onCancel);
|
| +}
|
| + bool get isBroadcast => true;
|
| + StreamSubscription<T> listen(void onData(T data), {
|
| +Function onError, void onDone(), bool cancelOnError}
|
| +) {
|
| +if (_controller == null || _controller.isClosed) {
|
| +return new _DoneStreamSubscription<T>(onDone);
|
| +}
|
| + if (_subscription == null) {
|
| +_subscription = _source.listen(_controller.add, onError: _controller.addError, onDone: _controller.close);
|
| +}
|
| + cancelOnError = identical(true, cancelOnError);
|
| + return _controller._subscribe(onData, onError, onDone, cancelOnError);
|
| +}
|
| + void _onCancel() {
|
| +bool shutdown = (_controller == null) || _controller.isClosed;
|
| + if (_onCancelHandler != null) {
|
| +_zone.runUnary(DDC$RT.wrap((void f(StreamSubscription<dynamic> __u112)) {
|
| +void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic, DDC$RT.type((StreamSubscription<dynamic> _) {
|
| +}
|
| +), "CastParam", """line 842, column 22 of dart:async/stream_impl.dart: """, x0 is StreamSubscription<dynamic>, true));
|
| + return f == null ? null : c;
|
| +}
|
| +, _onCancelHandler, __t109, __t100, "Wrap", """line 842, column 22 of dart:async/stream_impl.dart: """, _onCancelHandler is __t100), new _BroadcastSubscriptionWrapper(this));
|
| +}
|
| + if (shutdown) {
|
| +if (_subscription != null) {
|
| +_subscription.cancel();
|
| + _subscription = null;
|
| +}
|
| +}
|
| +}
|
| + void _onListen() {
|
| +if (_onListenHandler != null) {
|
| +_zone.runUnary(DDC$RT.wrap((void f(StreamSubscription<dynamic> __u113)) {
|
| +void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic, DDC$RT.type((StreamSubscription<dynamic> _) {
|
| +}
|
| +), "CastParam", """line 854, column 22 of dart:async/stream_impl.dart: """, x0 is StreamSubscription<dynamic>, true));
|
| + return f == null ? null : c;
|
| +}
|
| +, _onListenHandler, __t109, __t100, "Wrap", """line 854, column 22 of dart:async/stream_impl.dart: """, _onListenHandler is __t100), new _BroadcastSubscriptionWrapper(this));
|
| +}
|
| +}
|
| + void _cancelSubscription() {
|
| +if (_subscription == null) return; StreamSubscription subscription = _subscription;
|
| + _subscription = null;
|
| + _controller = null;
|
| + subscription.cancel();
|
| +}
|
| + void _pauseSubscription(Future resumeSignal) {
|
| +if (_subscription == null) return; _subscription.pause(resumeSignal);
|
| +}
|
| + void _resumeSubscription() {
|
| +if (_subscription == null) return; _subscription.resume();
|
| +}
|
| + bool get _isSubscriptionPaused {
|
| +if (_subscription == null) return false;
|
| + return _subscription.isPaused;
|
| +}
|
| +}
|
| + class _BroadcastSubscriptionWrapper<T> implements StreamSubscription<T> {final _AsBroadcastStream _stream;
|
| + _BroadcastSubscriptionWrapper(this._stream);
|
| + void onData(void handleData(T data)) {
|
| +throw new UnsupportedError("Cannot change handlers of asBroadcastStream source subscription.");
|
| +}
|
| + void onError(void handleError(Object data)) {
|
| +throw new UnsupportedError("Cannot change handlers of asBroadcastStream source subscription.");
|
| +}
|
| + void onDone(void handleDone()) {
|
| +throw new UnsupportedError("Cannot change handlers of asBroadcastStream source subscription.");
|
| +}
|
| + void pause([Future resumeSignal]) {
|
| +_stream._pauseSubscription(resumeSignal);
|
| +}
|
| + void resume() {
|
| +_stream._resumeSubscription();
|
| +}
|
| + Future cancel() {
|
| +_stream._cancelSubscription();
|
| + return null;
|
| +}
|
| + bool get isPaused {
|
| +return _stream._isSubscriptionPaused;
|
| +}
|
| + Future asFuture([var futureValue]) {
|
| +throw new UnsupportedError("Cannot change handlers of asBroadcastStream source subscription.");
|
| +}
|
| +}
|
| + class _StreamIteratorImpl<T> implements StreamIterator<T> {static const int _STATE_FOUND = 0;
|
| + static const int _STATE_DONE = 1;
|
| + static const int _STATE_MOVING = 2;
|
| + static const int _STATE_EXTRA_DATA = 3;
|
| + static const int _STATE_EXTRA_ERROR = 4;
|
| + static const int _STATE_EXTRA_DONE = 5;
|
| + StreamSubscription _subscription;
|
| + T _current = ((__x114) => DDC$RT.cast(__x114, Null, T, "CastLiteral", """line 968, column 16 of dart:async/stream_impl.dart: """, __x114 is T, false))(null);
|
| + var _futureOrPrefetch = null;
|
| + int _state = _STATE_FOUND;
|
| + _StreamIteratorImpl(final Stream<T> stream) {
|
| +_subscription = stream.listen(_onData, onError: _onError, onDone: _onDone, cancelOnError: true);
|
| +}
|
| + T get current => _current;
|
| + Future<bool> moveNext() {
|
| +if (_state == _STATE_DONE) {
|
| +return new _Future<bool>.immediate(false);
|
| +}
|
| + if (_state == _STATE_MOVING) {
|
| +throw new StateError("Already waiting for next.");
|
| +}
|
| + if (_state == _STATE_FOUND) {
|
| +_state = _STATE_MOVING;
|
| + _current = ((__x115) => DDC$RT.cast(__x115, Null, T, "CastLiteral", """line 998, column 18 of dart:async/stream_impl.dart: """, __x115 is T, false))(null);
|
| + _futureOrPrefetch = new _Future<bool>();
|
| + return DDC$RT.cast(_futureOrPrefetch, dynamic, DDC$RT.type((Future<bool> _) {
|
| +}
|
| +), "CastGeneral", """line 1000, column 14 of dart:async/stream_impl.dart: """, _futureOrPrefetch is Future<bool>, false);
|
| +}
|
| + else {
|
| +assert (_state >= _STATE_EXTRA_DATA); switch (_state) {case _STATE_EXTRA_DATA: _state = _STATE_FOUND;
|
| + _current = DDC$RT.cast(_futureOrPrefetch, dynamic, T, "CastGeneral", """line 1006, column 22 of dart:async/stream_impl.dart: """, _futureOrPrefetch is T, false);
|
| + _futureOrPrefetch = null;
|
| + _subscription.resume();
|
| + return new _Future<bool>.immediate(true);
|
| + case _STATE_EXTRA_ERROR: AsyncError prefetch = DDC$RT.cast(_futureOrPrefetch, dynamic, AsyncError, "CastGeneral", """line 1011, column 33 of dart:async/stream_impl.dart: """, _futureOrPrefetch is AsyncError, true);
|
| + _clear();
|
| + return new _Future<bool>.immediateError(prefetch.error, prefetch.stackTrace);
|
| + case _STATE_EXTRA_DONE: _clear();
|
| + return new _Future<bool>.immediate(false);
|
| +}
|
| +}
|
| +}
|
| + void _clear() {
|
| +_subscription = null;
|
| + _futureOrPrefetch = null;
|
| + _current = ((__x116) => DDC$RT.cast(__x116, Null, T, "CastLiteral", """line 1026, column 16 of dart:async/stream_impl.dart: """, __x116 is T, false))(null);
|
| + _state = _STATE_DONE;
|
| +}
|
| + Future cancel() {
|
| +StreamSubscription subscription = _subscription;
|
| + if (_state == _STATE_MOVING) {
|
| +_Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic, DDC$RT.type((_Future<bool> _) {
|
| +}
|
| +), "CastGeneral", """line 1033, column 31 of dart:async/stream_impl.dart: """, _futureOrPrefetch is _Future<bool>, false);
|
| + _clear();
|
| + hasNext._complete(false);
|
| +}
|
| + else {
|
| +_clear();
|
| +}
|
| + return subscription.cancel();
|
| +}
|
| + void _onData(T data) {
|
| +if (_state == _STATE_MOVING) {
|
| +_current = data;
|
| + _Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic, DDC$RT.type((_Future<bool> _) {
|
| +}
|
| +), "CastGeneral", """line 1045, column 31 of dart:async/stream_impl.dart: """, _futureOrPrefetch is _Future<bool>, false);
|
| + _futureOrPrefetch = null;
|
| + _state = _STATE_FOUND;
|
| + hasNext._complete(true);
|
| + return;}
|
| + _subscription.pause();
|
| + assert (_futureOrPrefetch == null); _futureOrPrefetch = data;
|
| + _state = _STATE_EXTRA_DATA;
|
| +}
|
| + void _onError(Object error, [StackTrace stackTrace]) {
|
| +if (_state == _STATE_MOVING) {
|
| +_Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic, DDC$RT.type((_Future<bool> _) {
|
| +}
|
| +), "CastGeneral", """line 1059, column 31 of dart:async/stream_impl.dart: """, _futureOrPrefetch is _Future<bool>, false);
|
| + _clear();
|
| + hasNext._completeError(error, stackTrace);
|
| + return;}
|
| + _subscription.pause();
|
| + assert (_futureOrPrefetch == null); _futureOrPrefetch = new AsyncError(error, stackTrace);
|
| + _state = _STATE_EXTRA_ERROR;
|
| +}
|
| + void _onDone() {
|
| +if (_state == _STATE_MOVING) {
|
| +_Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic, DDC$RT.type((_Future<bool> _) {
|
| +}
|
| +), "CastGeneral", """line 1073, column 31 of dart:async/stream_impl.dart: """, _futureOrPrefetch is _Future<bool>, false);
|
| + _clear();
|
| + hasNext._complete(false);
|
| + return;}
|
| + _subscription.pause();
|
| + _futureOrPrefetch = null;
|
| + _state = _STATE_EXTRA_DONE;
|
| +}
|
| +}
|
| + typedef dynamic __t100(dynamic __u101);
|
| + typedef void __t102<T>(T __u103);
|
| + typedef dynamic __t105(dynamic __u106, dynamic __u107);
|
| + typedef void __t109(StreamSubscription<dynamic> __u110);
|
|
|