Redo StreamTransformers so they work with Stack traces.

sdk/lib/async/stream_impl.dart

Index: sdk/lib/async/stream_impl.dart
diff --git a/sdk/lib/async/stream_impl.dart b/sdk/lib/async/stream_impl.dart
index 54596a6b0e69d9daf43e9dfde31d535762a4ba0e..860f7d84267113fb02b0def0b5847559bbe184cd 100644
--- a/sdk/lib/async/stream_impl.dart
+++ b/sdk/lib/async/stream_impl.dart
@@ -36,6 +36,7 @@ abstract class _EventDispatch<T> {
  *
  * 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.
@@ -43,6 +44,7 @@ abstract class _EventDispatch<T> {
  *     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.
@@ -92,25 +94,14 @@ class _BufferingStreamSubscription<T> implements StreamSubscription<T>,
    */
   _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);
   }
 
   /**
@@ -149,6 +140,14 @@ class _BufferingStreamSubscription<T> implements StreamSubscription<T>,
     _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);