Add an [immediate] argument to 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 200 matching lines @@
     if (!_isClosed) {
       var request = new _TakeRequest<T>(count);
       _addRequest(request);
       return request.future;
     }
     throw _failClosed();
   }
 
   /// Cancels the underlying stream subscription.
   ///
-  /// The cancel operation waits until all previously requested
-  /// events have been processed, then it cancels the subscription
-  /// providing the events.
+  /// If [immediate] is `false` (the default), the cancel operation waits until
+  /// all previously requested events have been processed, then it cancels the
+  /// subscription providing the events.
+  ///
+  /// If [immediate] is `true`, the subscription is instead canceled
+  /// immediately. Any pending events will complete as though the stream had

[floitsch, 2015/07/09 09:47:52]

Could you please rewrite the "Any pending events ..." sentence. I had a hard
time understanding it.

Also don't use future tense.

[nweiz, 2015/07/09 19:43:36]

Can you say more about what you didn't understand?
Done in a separate CL.

[floitsch, 2015/07/13 09:39:38]

Any pending events complete with a 'closed'-event, as though the stream had
closed by itself.

[nweiz, 2015/07/13 20:07:21]

Right, but what about that sentence is confusing?

+  /// closed when [cancel] was called.
   ///
   /// 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() {
-    if (!_isClosed) {
-      _isClosed = true;
+  Future cancel({bool immediate: false}) {
+    if (_isClosed) throw _failClosed();
+    _isClosed = true;
+
+    if (!immediate) {
       var request = new _CancelRequest(this);
       _addRequest(request);
       return request.future;
     }
-    throw _failClosed();
+
+    if (_isDone) return new Future.value();
+    if (_subscription == null) _subscription = _sourceStream.listen(null);
+    var future = _subscription.cancel();
+    _onDone();
+    return future;
   }

[Lasse Reichstein Nielsen, 2015/07/08 07:49:22]

This is really two different functions dispatched on a boolean - all they share
is the first two lines.

Maybe it should rather be a
  Future cancelImmediate();
method on the side? I think I would prefer that.

[nweiz, 2015/07/09 01:03:19]

I considered that, but ended up going this direction because of the similarity
to cancel({force}) elsewhere. If you think a separate method is better, I'll
change it.

[floitsch, 2015/07/09 09:47:52]

One function looks fine to me.

 
   /// 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() {
     return new StateError("Already cancelled");
   }
 
   // Callbacks receiving the events of the source stream.
@@ skipping 23 matching lines @@
       if (!request.addEvents(_eventQueue)) {
         request.close(_eventQueue);
       }
       return;
     }
     if (_requestQueue.isEmpty) {
       if (request.addEvents(_eventQueue)) return;
       _ensureListening();
     }
     _requestQueue.add(request);
-
   }
 
   /// Ensures that we are listening on events from [_sourceStream].
   ///
   /// Resumes subscription on [_sourceStream], or creates it if necessary.
   void _ensureListening() {
     assert(!_isDone);
     if (_subscription == null) {
       _subscription =
           _sourceStream.listen(_onData, onError: _onError, onDone: _onDone);
@@ skipping 316 matching lines @@
       _completer.complete(true);
       return true;
     }
     return false;
   }
 
   void close(_) {
     _completer.complete(false);
   }
 }