Avoid uncaught errors inside user code in stream-controller callbacks.

sdk/lib/async/stream_controller.dart

Index: sdk/lib/async/stream_controller.dart
diff --git a/sdk/lib/async/stream_controller.dart b/sdk/lib/async/stream_controller.dart
index 9ea10f52cb35788d7578e7fe6203d9fc4095b202..a9f5e71d7db2803bbbf2d846727a7a55825c2583 100644
--- a/sdk/lib/async/stream_controller.dart
+++ b/sdk/lib/async/stream_controller.dart
@@ -148,11 +148,23 @@ class _MultiControllerStream<T> extends _MultiStreamImpl<T> {
   _MultiControllerStream(this._subscriptionHandler, this._pauseHandler);
 
   void _onSubscriptionStateChange() {
-    if (_subscriptionHandler != null) _subscriptionHandler();
+    if (_subscriptionHandler != null) {
+      try {
+        _subscriptionHandler();
+      } catch (e, s) {
+        new AsyncError(e, s).throwDelayed();
+      }
+    }
   }
 
   void _onPauseStateChange() {
-    if (_pauseHandler != null) _pauseHandler();
+    if (_pauseHandler != null) {
+      try {
+        _pauseHandler();
+      } catch (e, s) {
+        new AsyncError(e, s).throwDelayed();
+      }
+    }
   }
 }
 
@@ -163,10 +175,22 @@ class _SingleControllerStream<T> extends _SingleStreamImpl<T> {
   _SingleControllerStream(this._subscriptionHandler, this._pauseHandler);
 
   void _onSubscriptionStateChange() {
-    if (_subscriptionHandler != null) _subscriptionHandler();
+    if (_subscriptionHandler != null) {
+      try {
+        _subscriptionHandler();
+      } catch (e, s) {
+        new AsyncError(e, s).throwDelayed();
+      }
+    }
   }
 
   void _onPauseStateChange() {
-    if (_pauseHandler != null) _pauseHandler();
+    if (_pauseHandler != null) {
+      try {
+        _pauseHandler();
+      } catch (e, s) {
+        new AsyncError(e, s).throwDelayed();
+      }
+    }
   }
 }