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..235418034f2ddd8c306e4ca696ca2e1ad2f8d7c4 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,48 @@ 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) {
+      Future cancelFuture;
+      try {

[Lasse Reichstein Nielsen, 2014/05/07 14:33:32]

This entire try-catch could just be `cancelFuture = new Future.sync(_onCancel);`
- if we don't mind always returning a future, even if _onCancel doesn't.

+        var cancelReturn = _onCancel();
+        if (cancelReturn is Future) {
+          cancelFuture = cancelReturn;
+        }
+      } catch (e, s) {
+        cancelFuture = new _Future.immediateError(e, s);
+      }
+
+      if (cancelFuture != null) {
+        if (result != null) {
+          result = result.whenComplete(() => cancelFuture);

[floitsch, 2014/05/07 15:16:25]

I think we should wait for the source-stream's cancel before invoking our
cancel.

That is (something similar to):

Future addingStreamCancelFuture;
if (_isAddingStream) {
  ...
  addingStreamCancelFuture = addState.cancel();
}

_varData = null;
_state = ...;

localCancel() {
  ...
}

if (addingStreamCancelFuture != null) {
  return addingStreamCancelFuture.whenComplete(localCancel);
}
return localCancel();

[Lasse Reichstein Nielsen, 2014/05/07 17:40:26]

Any reason for wanting to wait?
We know we are going to do a cleanup, so why not start as soon as possible,
instead of waiting for something else (and probably unrelated) to stop first?
We will still get the errors in the same order, it's just that we allow our
async operations to interleave, which is what async is good at.

[floitsch, 2014/05/07 17:50:50]

Let's discuss this tomorrow.
I don't have a good argument right now, but it feels safer if we wait.

+        } else {
+          result = cancelFuture;
+        }
+      }
+    }
+
     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.whenComplete(complete);
     } else {
       complete();
     }
-    return future;
+
+    return result;
   }
 
   void _recordPause(StreamSubscription<T> subscription) {
@@ -704,8 +723,7 @@ class _AddStreamState<T> {
   }
 
   void cancel() {
-    addSubscription.cancel();
-    complete();
+    addStreamFuture._asyncComplete(addSubscription.cancel());
   }
 
   void complete() {