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

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 402 matching lines @@
    * After closing, no further events may be added using [add] or [addError].
    *
    * You are allowed to close the controller more than once, but only the first
    * call has any effect.
    *
    * The first time a controller is closed, a "done" event is sent to its
    * stream.
    */
   Future close() {
     if (isClosed) {
-      _ensureDoneFuture();
-      return _doneFuture;
+      return _ensureDoneFuture();
     }
     if (!_mayAddEvent) throw _badEventState();
     _state |= _STATE_CLOSED;
     if (hasListener) {
       _sendDone();
     } else if (_isInitialState) {
       _ensurePendingEvents().add(const _DelayedDone());
     }
-    _ensureDoneFuture();
-    return _doneFuture;
+    return _ensureDoneFuture();
   }
 
   // EventSink interface. Used by the [addStream] events.
 
   // Add data event, used both by the [addStream] events and by [add].
   void _add(T value) {
     if (hasListener) {
       _sendData(value);
     } else if (_isInitialState) {
       _ensurePendingEvents().add(new _DelayedData<T>(value));
@@ skipping 37 matching lines @@
     }
     subscription._setPendingEvents(pendingEvents);
     subscription._guardCallback(() {
       _runGuarded(_onListen);
     });
 
     return subscription;
   }
 
   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) {
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       addState.pause();
     }
     _runGuarded(_onPause);
   }
 
@@ skipping 173 matching lines @@
       };
 
   void pause() {
     addSubscription.pause();
   }
 
   void resume() {
     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() {
     addStreamFuture._asyncComplete(null);
   }
 }
 
 class _StreamControllerAddStreamState<T> extends _AddStreamState<T> {
   // The subscription or pending data of a _StreamController.
   // Stored here because we reuse the `_varData` field  in the _StreamController
   // to store this state object.
   var varData;
 
   _StreamControllerAddStreamState(_StreamController controller,
                                   this.varData,
                                   Stream source,
                                   bool cancelOnError)
       : super(controller, source, cancelOnError) {
     if (controller.isPaused) {
       addSubscription.pause();
     }
   }
 }