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

tests/lib/async/stream_controller_test.dart

 // Copyright (c) 2011, 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.
 
 // Test the basic StreamController and StreamController.singleSubscription.
 library stream_controller_test;
 
 import "package:expect/expect.dart";
+import "package:async_helper/async_helper.dart";
 import 'dart:async';
 import 'event_helper.dart';
 
+fail(e) { Expect.fail("Unexepected error: $e"); }
+
 void testMultiController() {
   // Test normal flow.
   var c = new StreamController(sync: true);
   Events expectedEvents = new Events()
       ..add(42)
       ..add("dibs")
       ..error("error!")
       ..error("error too!")
       ..close();
   CaptureEvents actualEvents = new Events.capture(c.stream.asBroadcastStream());
@@ skipping 407 matching lines @@
   c = new StreamController.broadcast(sync: false);
   Expect.equals(c.stream, c.stream);
   c = new StreamController.broadcast(sync: true);
   Expect.equals(c.stream, c.stream);
   c = new StreamController.broadcast(sync: false, onListen:(){});
   Expect.equals(c.stream, c.stream);
   c = new StreamController.broadcast(sync: true, onListen:(){});
   Expect.equals(c.stream, c.stream);
 }
 
+void testCancelThrow() {
+  asyncStart();
+  asyncStart();
+  StreamController c = new StreamController(onCancel: () {
+    asyncEnd();
+    throw "ERROR";
+  });
+  c.add(1);
+  c.add(2);
+  c.add(3);
+  Future done = c.close();
+  StreamSubscription sub;
+  sub = c.stream.listen((v) {
+    Expect.equals(1, v);
+    Future f = sub.cancel();
+    f.catchError((e) {
+      // Must complete with error from onCancel.
+      Expect.equals("ERROR", e);
+      asyncEnd();
+    });
+  });
+  done.catchError(fail).whenComplete(asyncEnd);  // Must complete without error.
+}
+
+void testCancelThrow2() {
+  asyncStart();
+  asyncStart();
+  asyncStart();
+  asyncStart();
+  asyncStart();
+  StreamController c2 = new StreamController(onCancel: () {
+    asyncEnd();
+    throw "ERROR";
+  });
+  c2.add(1);
+  c2.add(2);
+  Future done2 = c2.close();
+  done2.catchError(fail).whenComplete(asyncEnd);  // Should not get error;
+
+  StreamController c = new StreamController();
+  var sub;
+  sub = c.stream.listen((v) {
+    Expect.equals(1, v);
+    Future f = sub.cancel();
+    f.catchError((e) {
+      // Error from addStream stream's cancel must go only here.
+      asyncEnd();
+      Expect.equals("ERROR", e);
+    });
+  });
+  var addDone = c.addStream(c2.stream);
+  addDone.catchError(fail).whenComplete(asyncEnd);  // Should not get error.
+  var done = c.done;
+  done.catchError(fail).whenComplete(asyncEnd);  // Should not get error.
+}
+
+void testCancelThrow3() {
+  asyncStart();
+  asyncStart();
+  asyncStart();
+  asyncStart();
+  asyncStart();
+  asyncStart();
+  StreamController c2 = new StreamController(onCancel: () {
+    asyneEnd();
+    throw "ERROR2";
+  });
+  c2.add(1);
+  c2.add(2);
+  var done2 = c2.close();
+  done2.catchError(fail).whenComplete(asyncEnd);  // Should not get error;
+
+  StreamController c = new StreamController(onCancel: () {
+    asyncEnd();
+    throw "ERROR1";
+  });
+  var sub;
+  sub = c.stream.listen((v) {
+    Expect.equals(1, v);
+    Future f = sub.cancel();
+    f.catchError((e) {
+      // Only the last error ends up here.
+      Expect.equals("ERROR1", e);
+      asyncEnd();
+    });
+  });
+  var addDone = c.addStream(c2.stream);
+  addDone.catchError(fail).whenComplete(asyncEnd);  // Error must not go here.
+  c.done.catchError(fail).whenComplete(asyncEnd); // Error must not go here.
+}
+
 main() {
   testMultiController();
   testSingleController();
   testExtraMethods();
   testClosed();
   testStreamEquals();
+  testCancelThrow();
+  testCancelThrow2();
+  testCancelThrow3();
 }