Redo StreamTransformers so they work with Stack traces.

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, StackTrace stackTrace);
@@ skipping 18 matching lines @@
  *
  * The only public methods are those of [StreamSubscription], so instances of
  * [_BufferingStreamSubscription] can be returned directly as a
  * [StreamSubscription] without exposing internal functionality.
  *
  * The [StreamController] is a public facing version of [Stream] and this class,
  * with some methods made public.
  *
  * The user interface of [_BufferingStreamSubscription] are the following
  * methods:
+ *
  * * [_add]: Add a data event to the stream.
  * * [_addError]: Add an error event to the stream.
  * * [_close]: Request to close the stream.
  * * [_onCancel]: Called when the subscription will provide no more events,
  *     either due to being actively canceled, or after sending a done event.
  * * [_onPause]: Called when the subscription wants the event source to pause.
  * * [_onResume]: Called when allowing new events after a pause.
+ *
  * The user should not add new events when the subscription requests a paused,
  * but if it happens anyway, the subscription will enqueue the events just as
  * when new events arrive while still firing an old event.
  */
 class _BufferingStreamSubscription<T> implements StreamSubscription<T>,
                                                  _EventSink<T>,
                                                  _EventDispatch<T> {
   /** The `cancelOnError` flag from the `listen` call. */
   static const int _STATE_CANCEL_ON_ERROR = 1;
   /**
@@ skipping 29 matching lines @@
   /** 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(void onData(T data),
-                               Function onError,
-                               void onDone(),
+  _BufferingStreamSubscription(void onDataHandler(T data),
+                               Function onErrorHandler,
+                               void onDoneHandler(),
                                bool cancelOnError)
-      : _onData = Zone.current.registerUnaryCallback(onData),
-        _onError = _registerErrorCallback(onError),
-        _onDone = Zone.current.registerCallback(onDone),
-        _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) {
-    assert(_onData != null);
-    assert(_onError != null);
-    assert(_onDone != null);
-  }
-
-  static _registerErrorCallback(Function errorCallback) {
-    if (errorCallback is ZoneBinaryCallback) {
-      return Zone.current.registerBinaryCallback(errorCallback);
-    } else {
-      return Zone.current.registerUnaryCallback(errorCallback);
-    }
+    : _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) {
+    onData(onDataHandler);
+    onError(onErrorHandler);
+    onDone(onDoneHandler);
   }
 
   /**
    * 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 18 matching lines @@
     return events;
   }
 
   // StreamSubscription interface.
 
   void onData(void handleData(T event)) {
     if (handleData == null) handleData = _nullDataHandler;
     _onData = Zone.current.registerUnaryCallback(handleData);
   }
 
+  static _registerErrorCallback(Function errorCallback) {
+    if (errorCallback is ZoneBinaryCallback) {
+      return Zone.current.registerBinaryCallback(errorCallback);
+    } else {
+      return Zone.current.registerUnaryCallback(errorCallback);
+    }
+  }
+
   void onError(Function handleError) {
     if (handleError == null) handleError = _nullErrorHandler;
     _onError = _registerErrorCallback(handleError);
   }
 
   void onDone(void handleDone()) {
     if (handleDone == null) handleDone = _nullDoneHandler;
     _onDone = Zone.current.registerCallback(handleDone);
   }
 
@@ skipping 844 matching lines @@
       _Future<bool> hasNext = _futureOrPrefetch;
       _clear();
       hasNext._complete(false);
       return;
     }
     _subscription.pause();
     _futureOrPrefetch = null;
     _state = _STATE_EXTRA_DONE;
   }
 }