Fix a bug in StreamQueue.cancel().

test/stream_queue_test.dart

 // Copyright (c) 2015, 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 filevents.
 
 import "dart:async";
 
 import "package:async/async.dart" show StreamQueue;
 import "package:test/test.dart";
 
 import "utils.dart";
@@ skipping 326 matching lines @@
     test("closes the events, prevents any other operation", () async {
       var events = new StreamQueue<int>(createStream());
       await events.cancel();
       expect(() => events.next, throwsStateError);
       expect(() => events.skip(1), throwsStateError);
       expect(() => events.take(1), throwsStateError);
       expect(() => events.rest, throwsStateError);
       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);
       var events = new StreamQueue<int>(controller.stream);
       controller.add(1);
       expect(await events.next, 1);
       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);
         expect(() => events.next, throwsStateError);
         expect(() => events.skip(1), throwsStateError);
         expect(() => events.take(1), throwsStateError);
         expect(() => events.rest, throwsStateError);
         expect(() => events.cancel(), throwsStateError);
       });
 
       test("cancels the underlying subscription immediately", () async {
         var controller = new StreamController();
         controller.add(1);
 
         var events = new StreamQueue<int>(controller.stream);
         expect(await events.next, 1);
         expect(controller.hasListener, isTrue);
 
         events.cancel(immediate: true);
         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);
         expect(events.next, throwsStateError);
         expect(events.hasNext, completion(isFalse));
 
         await events.cancel(immediate: true);
       });
 
       test("returns the result of closing the underlying subscription",
@@ skipping 288 matching lines @@
     controller.add(4);
     await flushMicrotasks();
     controller.close();
   }();
   return controller.stream;
 }
 
 Stream<int> createLongStream(int eventCount) async* {
   for (int i = 0; i < eventCount; i++) yield i;
 }