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..34582ae7a27603b74dcc2e17ccb64f2c1017593b 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,54 @@ abstract class _StreamController<T> implements StreamController<T>,
   }
 
   Future _recordCancel(StreamSubscription<T> subscription) {
+    // When we cancel, we first cancel any stream being added,
+    // Then we call _onCancel, and finally the _doneFuture is completed.
+    // If either of addStream's cancel or _onCancel returns a future,
+    // we wait for it before continuing.
+    // Any error during this process ends up in the returned future.
+    // If more errors happen, we act as if it happens inside nested try/finallys
+    // or whenComplete calls, and only the last error ends up in the
+    // returned future.
+    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) {
+      if (result == null) {
+        // Only introduce a future if one is needed.
+        // If _onCancel returns null, no future is needed.
+        try {
+          result = _onCancel();
+        } catch (e, s) {
+          // Return the error in the returned future.
+          // Complete it asynchronously, so there is time for a listener
+          // to handle the error.
+          result = new _Future().._asyncCompleteError(e, s);
+        }
+      } else {
+        // Simpler case when we already know that we will return a future.
+        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 +728,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() {