sdk/lib/async/stream_impl.dart - Issue 23875032: Expose Zones.

Unified Diff: sdk/lib/async/stream_impl.dart

Issue 23875032: Expose Zones. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart

Patch Set: Mark stack trace test as failing. Created 7 years, 3 months ago

Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.

Jump to:

View side-by-side diff with in-line comments

Index: sdk/lib/async/stream_impl.dart

diff --git a/sdk/lib/async/stream_impl.dart b/sdk/lib/async/stream_impl.dart

index ffab312f3c68c27e2e0f8d2b87190e71ed9b259d..8844d3ea800bd82571d51238659bfddd5ff0f128 100644

--- a/sdk/lib/async/stream_impl.dart

+++ b/sdk/lib/async/stream_impl.dart

@@ -80,7 +80,7 @@ class _BufferingStreamSubscription<T> implements StreamSubscription<T>,

_DataHandler<T> _onData;

_ErrorHandler _onError;

_DoneHandler _onDone;

- final _Zone _zone = _Zone.current;

+ final Zone _zone = Zone.current;

/** Bit vector based on state-constants above. */

int _state;

@@ -92,15 +92,17 @@ class _BufferingStreamSubscription<T> implements StreamSubscription<T>,

_PendingEvents _pending;

- _BufferingStreamSubscription(this._onData,

- this._onError,

- this._onDone,

+ _BufferingStreamSubscription(void onData(T data),

+ void onError(error),

+ void onDone(),

bool cancelOnError)

- : _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) {

+ : _onData = Zone.current.registerCallback1(onData),

+ _onError = Zone.current.registerCallback1(onError),

+ _onDone = Zone.current.registerCallback(onDone),

+ _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) {

assert(_onData != null);

assert(_onError != null);

assert(_onDone != null);

- _zone.expectCallback();

}

/**

@@ -219,7 +221,6 @@ class _BufferingStreamSubscription<T> implements StreamSubscription<T>,

void _cancel() {

_state |= _STATE_CANCELED;

- _zone.cancelCallbackExpectation();

if (_hasPending) {

_pending.cancelSchedule();

}

@@ -322,7 +323,7 @@ class _BufferingStreamSubscription<T> implements StreamSubscription<T>,

assert(!_inCallback);

bool wasInputPaused = _isInputPaused;

_state |= _STATE_IN_CALLBACK;

- _zone.executePeriodicCallbackGuarded(() => _onData(data));

+ _zone.runGuarded1(_onData, data);

Lasse Reichstein Nielsen 2013/09/23 14:24:12 Is that runGuarded*digit one*? It's very hard to r

floitsch 2013/09/23 17:12:07 Done.

_state &= ~_STATE_IN_CALLBACK;

_checkState(wasInputPaused);

}

@@ -333,11 +334,11 @@ class _BufferingStreamSubscription<T> implements StreamSubscription<T>,

assert(!_inCallback);

bool wasInputPaused = _isInputPaused;

_state |= _STATE_IN_CALLBACK;

- if (!_zone.inSameErrorZone(_Zone.current)) {

+ if (!_zone.inSameErrorZone(Zone.current)) {

// Errors are not allowed to traverse zone boundaries.

- _Zone.current.handleUncaughtError(error);

+ Zone.current.handleUncaughtError(error);

} else {

- _zone.executePeriodicCallbackGuarded(() => _onError(error));

+ _zone.runGuarded1(_onError, error);

Lasse Reichstein Nielsen 2013/09/23 14:24:12 Ditto.

floitsch 2013/09/23 17:12:07 Done.

}

_state &= ~_STATE_IN_CALLBACK;

if (_cancelOnError) {

@@ -351,7 +352,7 @@ class _BufferingStreamSubscription<T> implements StreamSubscription<T>,

assert(!_isPaused);

assert(!_inCallback);

_state |= (_STATE_CANCELED | _STATE_CLOSED | _STATE_IN_CALLBACK);

- _zone.executeCallbackGuarded(_onDone);

+ _zone.runGuarded(_onDone);

_onCancel(); // No checkState after cancel, it is always the last event.

_state &= ~_STATE_IN_CALLBACK;

}

@@ -532,7 +533,7 @@ void _nullDataHandler(var value) {}

/** Default error handler, reports the error to the global handler. */

void _nullErrorHandler(error) {

- _Zone.current.handleUncaughtError(error);

+ Zone.current.handleUncaughtError(error);

}

/** Default done handler, does nothing. */

@@ -723,18 +724,18 @@ class _AsBroadcastStream<T> extends Stream<T> {

final Stream<T> _source;

final _broadcastCallback _onListenHandler;

final _broadcastCallback _onCancelHandler;

- final _Zone _zone;

+ final Zone _zone;

_AsBroadcastStreamController<T> _controller;

StreamSubscription<T> _subscription;

_AsBroadcastStream(this._source,

- this._onListenHandler,

- this._onCancelHandler)

- : _zone = _Zone.current {

+ void onListenHandler(StreamSubscription subscription),

+ void onCancelHandler(StreamSubscription subscription))

+ : _onListenHandler = Zone.current.registerCallback1(onListenHandler),

+ _onCancelHandler = Zone.current.registerCallback1(onCancelHandler),

+ _zone = Zone.current {

_controller = new _AsBroadcastStreamController<T>(_onListen, _onCancel);

- // Keep zone alive until we are done doing callbacks.

- _zone.expectCallback();

}

bool get isBroadcast => true;

@@ -763,22 +764,19 @@ class _AsBroadcastStream<T> extends Stream<T> {

void _onCancel() {

bool shutdown = (_controller == null) || _controller.isClosed;

if (_onCancelHandler != null) {

- _zone.executePeriodicCallbackGuarded(

- () => _onCancelHandler(new _BroadcastSubscriptionWrapper(this)));

+ _zone.run1(_onCancelHandler, new _BroadcastSubscriptionWrapper(this));

Lasse Reichstein Nielsen 2013/09/23 14:24:12 runUnary?

floitsch 2013/09/23 17:12:07 Done.

}

if (shutdown) {

if (_subscription != null) {

_subscription.cancel();

_subscription = null;

}

- _zone.cancelCallbackExpectation();

}

void _onListen() {

if (_onListenHandler != null) {

- _zone.executePeriodicCallbackGuarded(

- () => _onListenHandler(new _BroadcastSubscriptionWrapper(this)));

+ _zone.run1(_onListenHandler, new _BroadcastSubscriptionWrapper(this));

}

@@ -788,9 +786,6 @@ class _AsBroadcastStream<T> extends Stream<T> {

// Called by [_controller] when it has no subscribers left.

StreamSubscription subscription = _subscription;

_subscription = null;

- if (_controller._isEmpty) {

- _zone.cancelCallbackExpectation();

- }

_controller = null; // Marks the stream as no longer listenable.

subscription.cancel();

}

« sdk/lib/async/stream_controller.dart ('K') | « sdk/lib/async/stream_controller.dart ('k') | sdk/lib/async/timer.dart » ('j') | sdk/lib/async/timer.dart » ('J')