Add zone support to streams.

sdk/lib/async/zone.dart

 // Copyright (c) 2013, 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;
 
 /**
  * A Zone represents the asynchronous version of a dynamic extent. Asynchronous
  * callbacks are executed in the zone they have been queued in. For example,
  * the callback of a `future.then` is executed in the same zone as the one where
@@ skipping 20 matching lines @@
    *
    * The main purpose of this method is to allow `this` to attach debugging
    * information to the returned zone.
    */
   _Zone fork();
 
   /**
    * Tells the zone that it needs to wait for one more callback before it is
    * done.
    *
-   * Use [executeCallback] or [unexpectCallback] when the callback is executed
+   * Use [executeCallback] or [cancelCallbackExpectation] when the callback is executed
    * (or canceled).
    */
   void expectCallback();
 
   /**
    * Tells the zone not to wait for a callback anymore.
    *
    * Prefer calling [executeCallback], instead. This method is mostly useful
    * for repeated callbacks (for example with [Timer.periodic]). In this case
    * one should should call [expectCallback] when the repeated callback is
-   * initiated, and [unexpectCallback] when the [Timer] is canceled.
+   * initiated, and [cancelCallbackExpectation] when the [Timer] is canceled.
    */
-  void unexpectCallback();
+  void cancelCallbackExpectation();
 
   /**
    * Executes the given callback in this zone.
    *
    * Decrements the number of callbacks this zone is waiting for (see
    * [expectCallback]).
    */
   void executeCallback(void fun());
 
   /**
    * Same as [executeCallback] but catches uncaught errors and gives them to
    * [handleUncaughtError].
    */
   void executeCallbackGuarded(void fun());
 
   /**
    * Same as [executeCallback] but does not decrement the number of
    * callbacks this zone is waiting for (see [expectCallback]).
    */
   void executePeriodicCallback(void fun());
 
   /**
    * Same as [executePeriodicCallback] but catches uncaught errors and gives
    * them to [handleUncaughtError].
    */
-  void executeGuardedPeriodicCallback(void fun());
+  void executePeriodicCallbackGuarded(void fun());
 
   /**
    * Runs [fun] asynchronously in this zone.
    */
   void runAsync(void fun());
 
   /**
    * Creates a Timer where the callback is executed in this zone.
    */
   Timer createTimer(Duration duration, void callback());
@@ skipping 33 matching lines @@
   final _Zone _parentZone;
 
   /// The children of this zone. A child's [_parentZone] is `this`.
   // TODO(floitsch): this should be a double-linked list.
   final List<_Zone> _children = <_Zone>[];
 
   /// The number of outstanding (asynchronous) callbacks. As long as the
   /// number is greater than 0 it means that the zone is not done yet.
   int _openCallbacks = 0;
 
+  bool _isExecutingCallback = false;
+
   _ZoneBase(this._parentZone) {
     _parentZone._addChild(this);
   }
 
   _ZoneBase._defaultZone() : _parentZone = null {
     assert(this is _DefaultZone);
   }
 
   _Zone get _errorZone => _parentZone._errorZone;
 
   void handleUncaughtError(error) {
     _parentZone.handleUncaughtError(error);
   }
 
   bool inSameErrorZone(_Zone otherZone) => _errorZone == otherZone._errorZone;
 
   _Zone fork() => this;
 
   expectCallback() => _openCallbacks++;
 
-  unexpectCallback() {
+  cancelCallbackExpectation() {
     _openCallbacks--;
     _checkIfDone();
   }
 
   /**
    * Cleans up this zone when it is done.
    *
    * This releases internal memore structures that are no longer necessary.
    *
    * A zone is done when its dynamic extent has finished executing and
    * there are no outstanding asynchronous callbacks.
    */
   _dispose() {
     if (_parentZone != null) {
       _parentZone._removeChild(this);
     }
   }
 
   /**
    * Checks if the zone is done and doesn't have any outstanding callbacks
    * anymore.
    *
    * This method is called when an operation has decremented the
    * outstanding-callback count, or when a child has been removed.
    */
   void _checkIfDone() {
-    if (_openCallbacks == 0 && _children.isEmpty) {
+    if (!_isExecutingCallback && _openCallbacks == 0 && _children.isEmpty) {
       _dispose();
     }
   }
 
   /**
    * Executes the given callback in this zone.
    *
    * Decrements the open-callback counter and checks (after the call) if the
    * zone is done.
    */
   void executeCallback(void fun()) {
     _openCallbacks--;
-    _runInZone(fun);
+    _runUnguarded(fun);

[Lasse Reichstein Nielsen, 2013/06/13 11:58:47]

That sounds like you could just have run fun directly,
Maybe write this._runUnguarded, to emphasize that it is the zone running it.

[floitsch, 2013/06/13 13:36:51]

Done.

   }
 
   /**
    * Same as [executeCallback] but catches uncaught errors and gives them to
    * [handleUncaughtError].
    */
   void executeCallbackGuarded(void fun()) {
     _openCallbacks--;
     _runGuarded(fun);
   }
 
   /**
    * Same as [executeCallback] but doesn't decrement the open-callback counter.
    */
   void executePeriodicCallback(void fun()) {
-    _runInZone(fun);
+    _runUnguarded(fun);
   }
 
   /**
    * Same as [executePeriodicCallback] but catches uncaught errors and gives
    * them to [handleUncaughtError].
    */
-  void executeGuardedPeriodicCallback(void fun()) {
+  void executePeriodicCallbackGuarded(void fun()) {
     _runGuarded(fun);
   }
 
-  _runInZone(fun()) {
-    if (identical(_Zone._current, this) && _openCallbacks != 0) return fun();
+  _runInZone(fun(), bool handleUncaught) {
+    if (identical(_Zone._current, this)
+        && !handleUncaught
+        && _isExecutingCallback) {
+      // No need to go through a try/catch.
+      return fun();
+    }
 
     _Zone oldZone = _Zone._current;
     _Zone._current = this;
     // While we are executing the function we don't want to have other
     // synchronous calls to think that they closed the zone. By incrementing
     // the _openCallbacks count we make sure that their test will fail.
     // As a side effect it will make nested calls faster since they are
     // (probably) in the same zone and have an _openCallbacks > 0.
-    _openCallbacks++;
+    bool oldIsExecuting = _isExecutingCallback;
+    _isExecutingCallback = true;

[Lasse Reichstein Nielsen, 2013/06/13 11:58:47]

So you can have nested callbacks?

[floitsch, 2013/06/13 13:36:51]

yes. For example a future-callback can invoke a sync-completer (which then has a
chain of futures).

     try {
       return fun();
+    } catch(e, s) {

[Lasse Reichstein Nielsen, 2013/06/13 11:58:47]

Seems overkill to catch and rethrow, but I guess it won't happen often.

[floitsch, 2013/06/13 13:36:51]

Otherwise I would need to duplicate the try/catch. Can be changed if it turns
out to be too expensive.

+      if (handleUncaught) {
+        handleUncaughtError(_asyncError(e, s));
+      } else {
+        rethrow;
+      }
     } finally {
-      _openCallbacks--;
+      _isExecutingCallback = oldIsExecuting;
       _Zone._current = oldZone;
       _checkIfDone();
     }
   }
 
   /**
    * Runs the function and catches uncaught errors.
    *
    * Uncaught errors are given to [handleUncaughtError].
    */
   _runGuarded(void fun()) {
-    try {
-      _runInZone(fun);
-    } catch(e, s) {
-      handleUncaughtError(_asyncError(e, s));
-    }
+    _runInZone(fun, true);
+  }
+
+  /**
+   * Runs the function but doesn't catch uncaught errors.
+   */
+  _runUnguarded(void fun()) {
+    _runInZone(fun, false);
   }
 
   runAsync(void fun()) {
     _openCallbacks++;
     _scheduleAsyncCallback(() {
       _openCallbacks--;
       _runGuarded(fun);
     });
   }
 
@@ skipping 45 matching lines @@
       var trace = getAttachedStackTrace(error);
       _attachStackTrace(error, null);
       if (trace != null) {
         print("Stack Trace:\n$trace\n");
       }
       throw error;
     });
   }
 }
 
+typedef void _CompletionCallback();
+
 /**
- * A zone that can execute a callback (through a future) when the zone is dead.
+ * A zone that executes a callback when the zone is dead.
  */
 class _WaitForCompletionZone extends _ZoneBase {
-  final Completer _doneCompleter = new Completer();
+  final _CompletionCallback _onDone;
 
-  _WaitForCompletionZone(_Zone parentZone) : super(parentZone);
+  _WaitForCompletionZone(_Zone parentZone, this._onDone) : super(parentZone);
 
   /**
-   * Runs the given function asynchronously and returns a future that is
-   * completed with `null` once the zone is done.
+   * Runs the given function asynchronously. Executes the [_onDone] callback
+   * when the zone is done.
    */
-  Future runWaitForCompletion(void fun()) {
-    _runInZone(() {
-      try {
-        fun();
-      } catch (e, s) {
-        handleUncaughtError(_asyncError(e, s));
-      }
-    });
-    return _doneCompleter.future;
+  void runWaitForCompletion(void fun()) {
+    this._runGuarded(fun);
   }
 
   _dispose() {
     super._dispose();
-    _doneCompleter.complete();
+    _onDone();
   }
 
   String toString() => "WaitForCompletion ${super.toString()}";
 }
 
+typedef bool _HandleErrorCallback(error);
+
 /**
  * A zone that collects all uncaught errors and provides them in a stream.
  * The stream is closed when the zone is done.
  */
 class _CatchErrorsZone extends _WaitForCompletionZone {
-  final StreamController errorsController = new StreamController();
+  final _HandleErrorCallback _handleError;
 
-  Stream get errors => errorsController.stream;
-
-  _CatchErrorsZone(_Zone parentZone) : super(parentZone);
+  _CatchErrorsZone(_Zone parentZone, this._handleError, void onDone())
+    : super(parentZone, onDone);
 
   _Zone get _errorZone => this;
 
   handleUncaughtError(error) {
-    errorsController.add(error);
-  }
-
-  Future runWaitForCompletion(void fun()) {
-    super.runWaitForCompletion(fun).whenComplete(() {
-      errorsController.close();
-    });
+    if (!_handleError(error)) _parentZone.handleUncaughtError(error);
   }
 
   String toString() => "WithErrors ${super.toString()}";
 }
 
 typedef void _TimerCallback();
 
 /**
  * A [Timer] class that takes zones into account.
  */
 class _ZoneTimer implements Timer {
   final _Zone _zone;
   final _TimerCallback _callback;
   Timer _timer;
   bool _isDone = false;
 
   _ZoneTimer(this._zone, Duration duration, this._callback) {
     _zone.expectCallback();
     _timer = _createTimer(duration, this.run);
   }
 
   void run() {
     _isDone = true;
     _zone.executeCallbackGuarded(_callback);
   }
 
   void cancel() {
-    if (!_isDone) _zone.unexpectCallback();
+    if (!_isDone) _zone.cancelCallbackExpectation();
     _isDone = true;
     _timer.cancel();
   }
 }
 
 typedef void _PeriodicTimerCallback(Timer timer);
 
 /**
  * A [Timer] class for periodic callbacks that takes zones into account.
  */
 class _PeriodicZoneTimer implements Timer {
   final _Zone _zone;
   final _PeriodicTimerCallback _callback;
   Timer _timer;
   bool _isDone = false;
 
   _PeriodicZoneTimer(this._zone, Duration duration, this._callback) {
     _zone.expectCallback();
     _timer = _createPeriodicTimer(duration, this.run);
   }
 
   void run(Timer timer) {
     assert(identical(_timer, timer));
-    _zone.executeGuardedPeriodicCallback(() { _callback(this); });
+    _zone.executePeriodicCallbackGuarded(() { _callback(this); });
   }
 
   void cancel() {
-    if (!_isDone) _zone.unexpectCallback();
+    if (!_isDone) _zone.cancelCallbackExpectation();
     _isDone = true;
     _timer.cancel();
   }
 }
 
 Stream catchErrors(void body()) {
-  _Zone catchErrorsZone = new _CatchErrorsZone(_Zone._current);
-  catchErrorsZone.runWaitForCompletion(body);
-  return catchErrorsZone.errors;
+  _CatchErrorsZone catchErrorsZone;
+  StreamController controller;
+
+  void onListen() {
+    catchErrorsZone.runWaitForCompletion(body);
+  }
+
+  bool handleError(e) {
+    controller.add(e);
+    return true;
+  }
+
+  void onDone() {
+    controller.close();
+  }
+
+  catchErrorsZone = new _CatchErrorsZone(_Zone._current, handleError, onDone);
+  controller = new StreamController(onListen: onListen);
+  return controller.stream;
 }
 
 Future waitForCompletion(void body()) {
-  _Zone zone = new _WaitForCompletionZone(_Zone._current);
-  return zone.runWaitForCompletion(body);
+  Completer completer = new Completer.sync();
+  _Zone zone = new _WaitForCompletionZone(_Zone._current, completer.complete);
+  zone.runWaitForCompletion(body);
+  return completer.future;
 }