Add zone support to streams.

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;
 
-/** Throws the given error in the next cycle. */
-_throwDelayed(var error, [Object stackTrace]) {
-  // We are going to reach the top-level here, but there might be a global
-  // exception handler. This means that we shouldn't print the stack trace.
-  // TODO(floitsch): Find better solution that doesn't print the stack trace
-  // if there is a global exception handler.
-  runAsync(() {
-    if (stackTrace != null) print(stackTrace);
-    var trace = getAttachedStackTrace(error);
-    if (trace != null && trace != stackTrace) print(trace);
-    throw error;
-  });
-}
-
 /** Abstract and private interface for a place to put events. */
 abstract class _EventSink<T> {
   void _add(T data);
   void _addError(Object error);
   void _close();
 }
 
 /**
  * Abstract and private interface for a place to send events.
  *
@@ skipping 58 matching lines @@
   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. */
   /* TODO(7733): Fix Function->_DataHandler<T> when dart2js understands
    * parameterized function types. */
   Function _onData;
   _ErrorHandler _onError;
   _DoneHandler _onDone;
+  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,
                                bool cancelOnError)
       : _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.unexpectCallback();

[Lasse Reichstein Nielsen, 2013/05/29 08:32:26]

Hmm, seeing it like this: unexpect is not a verb. It's stopExpecting, or
dontExpect.
Or other antonyms of expect: abandon, forfeit, forfend, shun :)

[floitsch, 2013/05/29 15:48:56]

cancelCallbackExpectation

     if (_hasPending) {
       _pending.cancelSchedule();
     }
   }
 
   /**
    * Increment the pause count.
    *
    * Also marks input as paused.
    */
@@ skipping 40 matching lines @@
     _state |= _STATE_CLOSED;
     if (_canFire) {
       _sendDone();
     } else {
       _addPending(const _DelayedDone());
     }
   }
 
   // Hooks called when the input is paused, unpaused or canceled.
   // These must not throw. If overwritten to call user code, include suitable
-  // try/catch wrapping and send any errors to [_throwDelayed].
+  // try/catch wrapping and send any errors to
+  // [_Zone.current.handleUncaughtError].
   void _onPause() {
     assert(_isInputPaused);
   }
 
   void _onResume() {
     assert(!_isInputPaused);
   }
 
   void _onCancel() {
     assert(_isCanceled);
@@ skipping 20 matching lines @@
   }
 
   /* _EventDispatch interface. */
 
   void _sendData(T data) {
     assert(!_isCanceled);
     assert(!_isPaused);
     assert(!_inCallback);
     bool wasInputPaused = _isInputPaused;
     _state |= _STATE_IN_CALLBACK;
-    try {
-      _onData(data);
-    } catch (e, s) {
-      _throwDelayed(e, s);
-    }
+    _zone.executeGuardedPeriodicCallback(() => _onData(data));
     _state &= ~_STATE_IN_CALLBACK;
     _checkState(wasInputPaused);
   }
 
   void _sendError(var error) {
     assert(!_isCanceled);
     assert(!_isPaused);
     assert(!_inCallback);
+    if (!_zone.inSameErrorZone(_Zone.current)) {
+      // Errors are not allowed to traverse zone boundaries.
+      _Zone.current.handleUncaughtError(error);
+      return;
+    }
     bool wasInputPaused = _isInputPaused;
     _state |= _STATE_IN_CALLBACK;
-    try {
-      _onError(error);
-    } catch (e, s) {
-      _throwDelayed(e, s);
-    }
+    _zone.executeGuardedPeriodicCallback(() => _onError(error));
     _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);
-    try {
-      _onDone();
-    } catch (e, s) {
-      _throwDelayed(e, s);
-    }
+    _zone.executeGuardedCallback(_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) {
-  _throwDelayed(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 648 matching lines @@
       _FutureImpl<bool> hasNext = _futureOrPrefetch;
       _clear();
       hasNext._setValue(false);
       return;
     }
     _subscription.pause();
     _futureOrPrefetch = null;
     _state = _STATE_EXTRA_DONE;
   }
 }