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 218 matching lines @@
   /// The returned future completes with the result of calling
   /// `cancel`.
   ///
   /// After calling `cancel`, no further events can be requested.
   /// None of [next], [rest], [skip], [take] or [cancel] may be
   /// called again.
   Future cancel({bool immediate: false}) {
     if (_isClosed) throw _failClosed();
     _isClosed = true;
 
+    if (_isDone) return new Future.value();
+    if (_subscription == null) _subscription = _sourceStream.listen(null);
+
     if (!immediate) {
       var request = new _CancelRequest(this);
       _addRequest(request);
       return request.future;

[Lasse Reichstein Nielsen, 2015/07/14 07:28:10]

That doesn't look right either - you subscribe to the stream, but don't set up
the callbacks. For a non-immediate cancel, that would mean it won't get the
onDone event, so the _CancelRequest never completes.

On the other hand, if you haven't subscribed yet, then it means that the cancel
is the first request, so the _eventQueue is empty.
In that case, you can just cancel it immediately, no need to enqueue anything.
So maybe change the enqueue-event case condition to:
 if (_isDone) return new Future.value();
 if (!immediate && _eventQueue.isNotEmpty) {
    assert(_subscription != null);
    ...
 }
 if (_subscription == null) _subscription = _sourceStream.listen(null);
 ...

Alternatively, andprobably better, fix this at the root cause: That a
_CancelRequest doesn't listen before it cancels if there is no subscription. See
below.

     }
 
-    if (_isDone) return new Future.value();
-    if (_subscription == null) _subscription = _sourceStream.listen(null);
     var future = _subscription.cancel();
     _onDone();
     return future;
   }
 
   /// Returns an error for when a request is made after cancel.
   ///
   /// Returns a [StateError] with a message saying that either
   /// [cancel] or [rest] have already been called.
   Error _failClosed() {
@@ skipping 267 matching lines @@
     _shutdown();
     return true;
   }
 
   void close(_) {
     _shutdown();
   }
 
   void _shutdown() {
     if (_streamQueue._subscription == null) {
       _completer.complete();

[Lasse Reichstein Nielsen, 2015/07/14 07:28:10]

This skips creating the subscription if it's null.
That's fine in the _isDone case, but not what you want in the not-subscribed-yet
case.
So change the condition here to `_streamQueue._isDone`, and add
`_streamQueue._ensureListening();` at the start of the else branch - or maybe a
litte more conservative:
 if (_streamQueue._subscription == null) _streamQueue._ensureListening();

[nweiz, 2015/07/14 20:14:08]

Done.

     } else {
       _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
@@ skipping 75 matching lines @@
       _completer.complete(true);
       return true;
     }
     return false;
   }
 
   void close(_) {
     _completer.complete(false);
   }
 }