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

sdk/lib/async/stream_controller.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;
 
 // -------------------------------------------------------------------
 // Controller for creating and adding events to a stream.
 // -------------------------------------------------------------------
 
@@ skipping 130 matching lines @@
 
 typedef void _NotificationHandler();
 
 class _MultiControllerStream<T> extends _MultiStreamImpl<T> {
   _NotificationHandler _subscriptionHandler;
   _NotificationHandler _pauseHandler;
 
   _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();
+      }
+    }
   }
 }
 
 class _SingleControllerStream<T> extends _SingleStreamImpl<T> {
   _NotificationHandler _subscriptionHandler;
   _NotificationHandler _pauseHandler;
 
   _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();
+      }
+    }
   }
 }