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) {
+    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) {
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       addState.pause();
     }
     _runGuarded(_onPause);
   }
 
@@ skipping 174 matching lines @@
 
   void pause() {
     addSubscription.pause();
   }
 
   void resume() {
     addSubscription.resume();
   }
 
   void cancel() {
-    addSubscription.cancel();
-    complete();
+    addStreamFuture._asyncComplete(addSubscription.cancel());
   }
 
   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();
     }
   }
 }