Use the async package's CancelableOperation.

lib/src/utils.dart

 // Copyright (c) 2013, 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 test.utils;
 
 import 'dart:async';
 import 'dart:convert';
 import 'dart:math' as math;
 
+import 'package:async/async.dart' hide StreamQueue;
 import 'package:crypto/crypto.dart';
 import 'package:path/path.dart' as p;
 import 'package:shelf/shelf.dart' as shelf;
 import 'package:stack_trace/stack_trace.dart';
 
 import 'backend/operating_system.dart';
-import 'util/cancelable_future.dart';
 import 'util/path_handler.dart';
 import 'util/stream_queue.dart';
 
 /// The maximum console line length.
 const _lineLength = 100;
 
 /// A typedef for a possibly-asynchronous function.
 ///
 /// The return type should only ever by [Future] or void.
 typedef AsyncFunction();
@@ skipping 288 matching lines @@
   }, onError: (error, stackTrace) {
     completer.completeError(error, stackTrace);
     subscription.cancel();
   }, onDone: () {
     completer.complete();
   });
 
   return completer.future;
 }
 
-/// Returns a [CancelableFuture] that returns the next value of [queue] unless
-/// it's canceled.
+/// Returns a [CancelableOperation] that returns the next value of [queue]
+/// unless it's canceled.
 ///
-/// If the future is canceled, [queue] is not moved forward at all. Note that
-/// it's not safe to call further methods on [queue] until this future has
+/// If the operation is canceled, [queue] is not moved forward at all. Note that
+/// it's not safe to call further methods on [queue] until this operation has
 /// either completed or been canceled.
-CancelableFuture cancelableNext(StreamQueue queue) {
+CancelableOperation cancelableNext(StreamQueue queue) {
   var fork = queue.fork();
-  var completer = new CancelableCompleter(() => fork.cancel(immediate: true));
+  var completer = new CancelableCompleter(
+      onCancel: () => fork.cancel(immediate: true));
   completer.complete(fork.next.then((_) {
     fork.cancel();
     return queue.next;
   }));
-  return completer.future;
+  return completer.operation;
 }
 
-/// Returns a single-subscription stream that emits the results of [futures] in
-/// the order they complete.
+/// Returns a single-subscription stream that emits the results of [operations]
+/// in the order they complete.
 ///
-/// If any futures in [futures] are [CancelableFuture]s, this will cancel them
-/// if the subscription is canceled.
-Stream inCompletionOrder(Iterable<Future> futures) {
-  var futureSet = futures.toSet();
+/// If the subscription is canceled, any pending operations are canceled as
+/// well.
+Stream inCompletionOrder(Iterable<CancelableOperation> operations) {
+  var operationSet = operations.toSet();
   var controller = new StreamController(sync: true, onCancel: () {
-    return Future.wait(futureSet.map((future) {
-      return future is CancelableFuture ? future.cancel() : null;
-    }).where((future) => future != null));
+    return Future.wait(operationSet.map((operation) => operation.cancel()));
   });
 
-  for (var future in futureSet) {
-    future.then(controller.add).catchError(controller.addError)
+  for (var operation in operationSet) {
+    operation.value.then(controller.add).catchError(controller.addError)
         .whenComplete(() {
-      futureSet.remove(future);
-      if (futureSet.isEmpty) controller.close();
+      operationSet.remove(operation);
+      if (operationSet.isEmpty) controller.close();
     });
   }
 
   return controller.stream;
 }
 
 /// Returns a stream that emits [error] and [stackTrace], then closes.
 ///
 /// This is useful for adding errors to streams defined via `async*`.
 Stream errorStream(error, StackTrace stackTrace) {
@@ skipping 26 matching lines @@
       urlSafe: urlSafe, addLineSeparator: addLineSeparator);
 }
 
 /// Returns middleware that nests all requests beneath the URL prefix [beneath].
 shelf.Middleware nestingMiddleware(String beneath) {
   return (handler) {
     var pathHandler = new PathHandler()..add(beneath, handler);
     return pathHandler.handler;
   };
 }