Fix a bug in StreamQueue.cancel().

lib/src/stream_queue.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 file.
 
 library async.stream_events;
 
 import 'dart:async';
 import 'dart:collection';
 
 import "subscription_stream.dart";
@@ skipping 512 matching lines @@
   bool addEvents(Queue<Result> events) {
     _shutdown();
     return true;
   }
 
   void close(_) {
     _shutdown();
   }
 
   void _shutdown() {
-    if (_streamQueue._subscription == null) {
+    if (_streamQueue._isDone) {
       _completer.complete();
     } else {
+      _streamQueue._ensureListening();
       _completer.complete(_streamQueue._dispose().cancel());
     }
   }
 }
 
 /// Request for a [StreamQueue.rest] call.
 ///
 /// The request is always complete, it just waits in the request queue
 /// until all previous events are fulfilled, then it takes over the
 /// stream events subscription and creates a stream from it.
@@ skipping 74 matching lines @@
       _completer.complete(true);
       return true;
     }
     return false;
   }
 
   void close(_) {
     _completer.complete(false);
   }
 }