Return futures on Stream.cancel when possible.

tests/lib/async/stream_subscription_cancel_test.dart

Index: tests/lib/async/stream_subscription_cancel_test.dart
diff --git a/tests/lib/async/stream_subscription_cancel_test.dart b/tests/lib/async/stream_subscription_cancel_test.dart
index d6c6987545fcc04568eaed5f881e8c08d8a37f94..48649b2717df0566aada0e9b0050562324910f44 100644
--- a/tests/lib/async/stream_subscription_cancel_test.dart
+++ b/tests/lib/async/stream_subscription_cancel_test.dart
@@ -27,17 +27,22 @@ void main() {
   test('subscription.cancel after close', () {
     var completer = new Completer();
     StreamController controller = new StreamController(
-        onCancel: completer.complete);
+        onCancel: () {
+          completer.complete();
+          return completer.future;
+        });
 
     controller.close();
 
+    var completer2 = new Completer();
     var sub;
     void onDone() {
-      expect(sub.cancel(), isNull);
+      sub.cancel().then(completer2.complete);
     }
 
     sub = controller.stream.listen(null, onDone: onDone);
     expect(completer.future, completes);
+    expect(completer2.future, completes);
   });
 
   test('subscription.cancel after error', () {
@@ -135,4 +140,183 @@ void main() {
       .cancel();
     expect(doneCompleter.future, completion(equals(true)));
   });
+
+  test('subscription.cancel through map', () {
+    var completer = new Completer();
+    StreamController controller = new StreamController(
+        onCancel: () => completer.future);
+
+    bool done = false;
+    var future = controller.stream.map((x) => x).listen(null).cancel();
+
+    expect(future.then((_) => done = true), completion(equals(true)));
+
+    Timer.run(() {
+      expect(done, isFalse);
+      completer.complete();
+    });
+  });
+
+  test('subscription.cancel through asyncMap', () {
+    var completer = new Completer();
+    StreamController controller = new StreamController(
+        onCancel: () => completer.future);
+
+    bool done = false;
+    var future = controller.stream.asyncMap((x) => x).listen(null).cancel();
+
+    expect(future.then((_) => done = true), completion(equals(true)));
+
+    Timer.run(() {
+      expect(done, isFalse);
+      completer.complete();
+    });
+  });
+
+  test('subscription.cancel through asyncExpand', () {
+    var completer = new Completer();
+    StreamController controller = new StreamController(
+        onCancel: () => completer.future);
+
+    bool done = false;
+    var future = controller.stream.asyncExpand((x) => x).listen(null).cancel();
+
+    expect(future.then((_) => done = true), completion(equals(true)));
+
+    Timer.run(() {
+      expect(done, isFalse);
+      completer.complete();
+    });
+  });
+
+  test('subscription.cancel through handleError', () {
+    var completer = new Completer();
+    StreamController controller = new StreamController(
+        onCancel: () => completer.future);
+
+    bool done = false;
+    var future = controller.stream.handleError((x) => x).listen(null).cancel();
+
+    expect(future.then((_) => done = true), completion(equals(true)));
+
+    Timer.run(() {
+      expect(done, isFalse);
+      completer.complete();
+    });
+  });
+
+  test('subscription.cancel through skip', () {
+    var completer = new Completer();
+    StreamController controller = new StreamController(
+        onCancel: () => completer.future);
+
+    bool done = false;
+    var future = controller.stream.skip(1).listen(null).cancel();
+
+    expect(future.then((_) => done = true), completion(equals(true)));
+
+    Timer.run(() {
+      expect(done, isFalse);
+      completer.complete();
+    });
+  });
+
+  test('subscription.cancel through take', () {
+    var completer = new Completer();
+    StreamController controller = new StreamController(
+        onCancel: () => completer.future);
+
+    bool done = false;
+    var future = controller.stream.take(1).listen(null).cancel();
+
+    expect(future.then((_) => done = true), completion(equals(true)));
+
+    Timer.run(() {
+      expect(done, isFalse);
+      completer.complete();
+    });
+  });
+
+  test('subscription.cancel through skipWhile', () {
+    var completer = new Completer();
+    StreamController controller = new StreamController(
+        onCancel: () => completer.future);
+
+    bool done = false;
+    var future = controller.stream.skipWhile((x) => true).listen(null).cancel();
+
+    expect(future.then((_) => done = true), completion(equals(true)));
+
+    Timer.run(() {
+      expect(done, isFalse);
+      completer.complete();
+    });
+  });
+
+  test('subscription.cancel through takeWhile', () {
+    var completer = new Completer();
+    StreamController controller = new StreamController(
+        onCancel: () => completer.future);
+
+    bool done = false;
+    var future = controller.stream.takeWhile((x) => true).listen(null).cancel();
+
+    expect(future.then((_) => done = true), completion(equals(true)));
+
+    Timer.run(() {
+      expect(done, isFalse);
+      completer.complete();
+    });
+  });
+
+  test('subscription.cancel through timeOut', () {
+    var completer = new Completer();
+    StreamController controller = new StreamController(
+        onCancel: () => completer.future);
+
+    bool done = false;
+    var duration = const Duration(hours: 5);
+    var future = controller.stream.timeout(duration).listen(null).cancel();
+
+    expect(future.then((_) => done = true), completion(equals(true)));
+
+    Timer.run(() {
+      expect(done, isFalse);
+      completer.complete();
+    });
+  });
+
+  test('subscription.cancel through transform', () {
+    var completer = new Completer();
+    StreamController controller = new StreamController(
+        onCancel: () => completer.future);
+
+    bool done = false;
+    var transformer =
+        new StreamTransformer.fromHandlers(handleData: (x, y) {});
+    var future = controller.stream.transform(transformer).listen(null).cancel();
+
+    expect(future.then((_) => done = true), completion(equals(true)));
+
+    Timer.run(() {
+      expect(done, isFalse);
+      completer.complete();
+    });
+  });
+
+  test('subscription.cancel through where', () {
+    var completer = new Completer();
+    StreamController controller = new StreamController(
+        onCancel: () => completer.future);
+
+    bool done = false;
+    var future = controller.stream.where((x) => true).listen(null).cancel();
+
+    expect(future.then((_) => done = true), completion(equals(true)));
+
+    Timer.run(() {
+      expect(done, isFalse);
+      completer.complete();
+    });
+  });
 }