Add collectBytesCancelable function.

lib/src/byte_collector.dart

Index: lib/src/byte_collector.dart
diff --git a/lib/src/byte_collector.dart b/lib/src/byte_collector.dart
index 3b4f075034f853275ab587434e1147568c3b47e2..a8d2dbf4a3e506c83e9ee00fed94404fe9155b9b 100644
--- a/lib/src/byte_collector.dart
+++ b/lib/src/byte_collector.dart
@@ -4,6 +4,7 @@
 
 import "dart:async";
 import "dart:typed_data";
+import "cancelable_operation.dart";
 
 /// Collects an asynchronous sequence of byte lists into a single list of bytes.
 ///
@@ -13,10 +14,39 @@ import "dart:typed_data";
 /// If any of the input data are not valid bytes, they will be truncated to
 /// an eight-bit unsigned value in the resulting list.
 Future<Uint8List> collectBytes(Stream<List<int>> source) {
-  var byteLists = List<List<int>>[];
+  return _collectBytes(source, (_, result) => result);
+}
+
+/// Collects an asynchronous sequence of byte lists into a single list of bytes.
+///
+/// Returns a [CancelableOperation] that provides the result future and a way
+/// to cancel the collection early.
+///
+/// If the [source] stream emits an error event,
+/// the collection fails and the returned future completes with the same error.
+///
+/// If any of the input data are not valid bytes, they will be truncated to
+/// an eight-bit unsigned value in the resulting list.
+CancelableOperation<Uint8List> collectBytesCancelable(
+    Stream<List<int>> source) {
+  return _collectBytes(source, (subscription, result) =>
+      new CancelableOperation.fromFuture(result, onCancel: subscription.cancel)
+  );
+}
+
+/// Generalization over [collectBytes] and [collectBytesCancelable].
+///
+/// Performs all the same operations, but the final result is created
+/// by the [result] function, which has access to the stream subscription
+/// so it can cancel the operation.
+T _collectBytes<T>(
+    Stream<List<int>> source,
+    T result(StreamSubscription<List<int>> subscription,
+        Future<Uint8List> result)) {
+  var byteLists = <List<int>>[];
   var length = 0;
   var completer = new Completer<Uint8List>.sync();
-  source.listen(
+  var subscription = source.listen(
       (bytes) {
         byteLists.add(bytes);
         length += bytes.length;
@@ -26,7 +56,7 @@ Future<Uint8List> collectBytes(Stream<List<int>> source) {
         completer.complete(_collect(length, byteLists));
       },
       cancelOnError: true);
-  return completer.future;
+  return result(subscription, completer.future);
 }
 
 // Join a lists of bytes with a known total length into a single [Uint8List].