Use a named param for StreamQueue.cancelImmediately.

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 73 matching lines @@
   ///
   /// Set to subscription when listening, and set to `null` when the
   /// subscription is done (and [_isDone] is set to true).
   StreamSubscription _subscription;
 
   /// Whether we have listened on [_sourceStream] and the subscription is done.
   bool _isDone = false;
 
   /// Whether a closing operation has been performed on the stream queue.
   ///
-  /// Closing operations are [cancel], [cancelImmediately], and [rest].
+  /// Closing operations are [cancel] and [rest].
   bool _isClosed = false;
 
   /// Queue of events not used by a request yet.
   final Queue<Result> _eventQueue = new Queue();
 
   /// Queue of pending requests.
   ///
   /// Access through methods below to ensure consistency.
   final Queue<_EventRequest> _requestQueue = new Queue();
 
@@ skipping 106 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 are completed as though the underlying
+  /// stream had closed.
   ///
   /// 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();
-  }
-
-  /// Cancels the underlying stream subscription immediately.
-  ///
-  /// Any pending events will complete as though the stream had closed when
-  /// [cancel] was called.
-  ///
-  /// The returned future completes with the result of calling
-  /// `StreamSubscription.cancel`.
-  ///
-  /// After calling `cancelImmediately`, no further events can be requested.
-  /// None of [next], [rest], [skip], [take] or [cancel] may be called again.
-  Future cancelImmediately() {
-    if (_isClosed) throw _failClosed();
-    _isClosed = true;
 
     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], [cancelImmediately], or [rest] have already been called.
+  /// [cancel] or [rest] have already been called.
   Error _failClosed() {
     return new StateError("Already cancelled");
   }
 
   // Callbacks receiving the events of the source stream.
 
   void _onData(T data) {
     _eventQueue.add(new Result.value(data));
     _checkQueues();
   }
@@ skipping 354 matching lines @@
       _completer.complete(true);
       return true;
     }
     return false;
   }
 
   void close(_) {
     _completer.complete(false);
   }
 }