Expose Zones.

sdk/lib/async/stream_impl.dart

 // 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;
 
 /** Abstract and private interface for a place to put events. */
 abstract class _EventSink<T> {
   void _add(T data);
   void _addError(Object error);
@@ skipping 62 matching lines @@
   static const int _STATE_CANCELED = 8;
   static const int _STATE_IN_CALLBACK = 16;
   static const int _STATE_HAS_PENDING = 32;
   static const int _STATE_PAUSE_COUNT = 64;
   static const int _STATE_PAUSE_COUNT_SHIFT = 6;
 
   /* Event handlers provided in constructor. */
   _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;
 
   /**
    * Queue of pending events.
    *
    * Is created when necessary, or set in constructor for preconfigured events.
    */
   _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();
   }
 
   /**
    * Sets the subscription's pending events object.
    *
    * This can only be done once. The pending events object is used for the
    * rest of the subscription's life cycle.
    */
   void _setPendingEvents(_PendingEvents pendingEvents) {
     assert(_pending == null);
@@ skipping 98 matching lines @@
   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;
-    _zone.cancelCallbackExpectation();
     if (_hasPending) {
       _pending.cancelSchedule();
     }
   }
 
   /**
    * Increment the pause count.
    *
    * Also marks input as paused.
    */
@@ skipping 82 matching lines @@
   }
 
   /* _EventDispatch interface. */
 
   void _sendData(T data) {
     assert(!_isCanceled);
     assert(!_isPaused);
     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 read (I read it as "lower case
l").
Zone method names don't have to be short, so how about runGuardedUnary?

[floitsch, 2013/09/23 17:12:07]

Done.

     _state &= ~_STATE_IN_CALLBACK;
     _checkState(wasInputPaused);
   }
 
   void _sendError(var error) {
     assert(!_isCanceled);
     assert(!_isPaused);
     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) {
       _cancel();
     }
     _checkState(wasInputPaused);
   }
 
   void _sendDone() {
     assert(!_isCanceled);
     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;
   }
 
   /**
    * Call a hook function.
    *
    * The call is properly wrapped in code to avoid other callbacks
    * during the call, and it checks for state changes after the call
    * that should cause further callbacks.
@@ skipping 160 matching lines @@
 typedef void _DataHandler<T>(T value);
 typedef void _ErrorHandler(error);
 typedef void _DoneHandler();
 
 
 /** Default data handler, does nothing. */
 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. */
 void _nullDoneHandler() {}
 
 
 /** A delayed event on a buffering stream subscription. */
 abstract class _DelayedEvent {
   /** Added as a linked list on the [StreamController]. */
   _DelayedEvent next;
@@ skipping 170 matching lines @@
   void cancel() {}
   bool get isPaused => _pauseCounter > 0;
 
   Future asFuture([futureValue]) => new _Future();
 }
 
 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;
 
   StreamSubscription<T> listen(void onData(T data),
                                { void onError(Object error),
                                  void onDone(),
                                  bool cancelOnError}) {
     if (_controller == null) {
       // Return a dummy subscription backed by nothing, since
       // it won't ever receive any events.
       return new _DummyStreamSubscription<T>();
     }
     if (_subscription == null) {
       _subscription = _source.listen(_controller.add,
                                      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);
   }
 
   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));
     }
   }
 
   // Methods called from _BroadcastSubscriptionWrapper.
   void _cancelSubscription() {
     if (_subscription == null) return;
     // 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();
   }
 
   void _pauseSubscription(Future resumeSignal) {
     if (_subscription == null) return;
     _subscription.pause(resumeSignal);
   }
 
   void _resumeSubscription() {
@@ skipping 196 matching lines @@
       _Future<bool> hasNext = _futureOrPrefetch;
       _clear();
       hasNext._complete(false);
       return;
     }
     _subscription.pause();
     _futureOrPrefetch = null;
     _state = _STATE_EXTRA_DONE;
   }
 }