Make errors from StreamController onCancel calls end up in the returned future.

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 d800825fbbb20dc41a6d4e48869a1df28c9c324e..244437da9c95fcef471d71e30e610819ce014bc4 100644
--- a/sdk/lib/async/stream_controller.dart
+++ b/sdk/lib/async/stream_controller.dart
@@ -420,8 +420,7 @@ abstract class _StreamController<T> implements StreamController<T>,
    */
   Future close() {
     if (isClosed) {
-      _ensureDoneFuture();
-      return _doneFuture;
+      return _ensureDoneFuture();
     }
     if (!_mayAddEvent) throw _badEventState();
     _state |= _STATE_CLOSED;
@@ -430,8 +429,7 @@ abstract class _StreamController<T> implements StreamController<T>,
     } else if (_isInitialState) {
       _ensurePendingEvents().add(const _DelayedDone());
     }
-    _ensureDoneFuture();
-    return _doneFuture;
+    return _ensureDoneFuture();
   }
 
   // EventSink interface. Used by the [addStream] events.
@@ -489,27 +487,40 @@ abstract class _StreamController<T> implements StreamController<T>,
   }
 
   Future _recordCancel(StreamSubscription<T> subscription) {
+    Future result;
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
-      addState.cancel();
+      result = addState.cancel();
     }
     _varData = null;
     _state =
         (_state & ~(_STATE_SUBSCRIBED | _STATE_ADDSTREAM)) | _STATE_CANCELED;
 
+    if (_onCancel != null) {

[floitsch, 2014/05/09 15:39:21]

Give general overview of how we want to handle errors.

+      if (result == null) {
+        try {
+          result = _onCancel();
+        } catch (e, s) {
+          result = new _Future().._asyncCompleteError(e, s);

[floitsch, 2014/05/09 15:39:21]

add comment why this is necessary.

+        }
+      } else {
+        result = result.whenComplete(_onCancel);
+      }
+    }
+
     void complete() {
       if (_doneFuture != null && _doneFuture._mayComplete) {
         _doneFuture._asyncComplete(null);
       }
     }
 
-    Future future = _runGuarded(_onCancel);
-    if (future != null) {
-      future = future.whenComplete(complete);
+    if (result != null) {
+      result = result.whenComplete(complete);
     } else {
       complete();
     }
-    return future;
+
+    return result;
   }
 
   void _recordPause(StreamSubscription<T> subscription) {
@@ -703,9 +714,21 @@ class _AddStreamState<T> {
     addSubscription.resume();
   }
 
-  void cancel() {
-    addSubscription.cancel();
-    complete();
+  /**
+   * Stop adding the stream.
+   *
+   * Complete the future returned by `StreamController.addStream` when
+   * the cancel is complete.
+   *
+   * Return a future if the cancel takes time, otherwise return `null`.
+   */
+  Future cancel() {
+    var cancel = addSubscription.cancel();
+    if (cancel == null) {
+      addStreamFuture._asyncComplete(null);
+      return null;
+    }
+    return cancel.whenComplete(() { addStreamFuture._asyncComplete(null); });
   }
 
   void complete() {