Fix a bug in StreamQueue.cancel().

test/stream_queue_test.dart

Index: test/stream_queue_test.dart
diff --git a/test/stream_queue_test.dart b/test/stream_queue_test.dart
index cad0ad511669f3f29ca9f44e4e639b31f57a35bd..228ba8adc4fb419a7c3b69d47c8494afb0d43005 100644
--- a/test/stream_queue_test.dart
+++ b/test/stream_queue_test.dart
@@ -344,6 +344,14 @@ main() {
       expect(() => events.cancel(), throwsStateError);
     });
 
+    test("cancels underlying subscription when called before any event",
+        () async {
+      var cancelFuture = new Future.value(42);
+      var controller = new StreamController(onCancel: () => cancelFuture);
+      var events = new StreamQueue<int>(controller.stream);
+      expect(await events.cancel(), 42);
+    });
+
     test("cancels underlying subscription, returns result", () async {
       var cancelFuture = new Future.value(42);
       var controller = new StreamController(onCancel: () => cancelFuture);
@@ -353,7 +361,7 @@ main() {
       expect(await events.cancel(), 42);
     });
 
-    group("with immediate: true", () async {
+    group("with immediate: true", () {
       test("closes the events, prevents any other operation", () async {
         var events = new StreamQueue<int>(createStream());
         await events.cancel(immediate: true);
@@ -376,6 +384,15 @@ main() {
         await expect(controller.hasListener, isFalse);
       });
 
+      test("cancels the underlying subscription when called before any event",
+          () async {
+        var cancelFuture = new Future.value(42);
+        var controller = new StreamController(onCancel: () => cancelFuture);
+
+        var events = new StreamQueue<int>(controller.stream);
+        expect(await events.cancel(immediate: true), 42);
+      });
+
       test("closes pending requests", () async {
         var events = new StreamQueue<int>(createStream());
         expect(await events.next, 1);