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..d85b1db2da9a3fc718cec13eabc55d95d8286198 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,50 @@ 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>>[];

[Lasse Reichstein Nielsen, 2017/01/27 12:20:48]

The type here was actually wrong (I just moved it from left to right without
removing the outer List), so this CL also fixes that bug.

+  return _collectBytes<Future<Uint8List>>(source, _futureResult);

[nweiz, 2017/01/27 21:56:00]

I think you shouldn't need an explicit generic type here (or below)—it should
infer it from the return type of the function, and from the callback as well.

If you do include the generic, either make it comment-style or update the SDK
constraint to have 1.21.0 as its minimum.

[Lasse Reichstein Nielsen, 2017/01/30 15:06:33]

It probably doesn't even need to be generic, I can just make it dynamic and do a
down-cast. Not sure what is better in the long run.

Will update to 1.21.0

[nweiz, 2017/01/30 22:21:26]

It does need to be generic in strong mode—but I believe it'll complain
statically if the generic type is missing or doesn't match up.

[Lasse Reichstein Nielsen, 2017/01/31 11:53:54]

I've removed all the explicit type declarations (even more now that the
functions are inline), and the analyzer is happy.

+}
+
+/// 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<CancelableOperation<Uint8List>>(
+      source, _cancelOperationResult);
+}
+
+// Helper function for [collectBytes].
+Future<Uint8List> _futureResult(_, Completer<Uint8List> completer) =>
+    completer.future;

[nweiz, 2017/01/27 21:56:00]

I'd make these inline—I think it reads clearer, and you get to avoid re-writing
a bunch of type annotations.

[Lasse Reichstein Nielsen, 2017/01/30 15:06:33]

Done.

+
+// Helper function for [collectBytesCancelable].
+CancelableOperation<Uint8List> _cancelOperationResult(
+    StreamSubscription<List<int>> subscription,
+    Completer<Uint8List> completer) {
+  return new CancelableOperation.fromFuture(completer.future,
+      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,
+        Completer<Uint8List> completer)) {

[nweiz, 2017/01/27 21:56:00]

I'd just pass the Future here, since that's the only thing that's actually used.

[Lasse Reichstein Nielsen, 2017/01/30 15:06:33]

True. Might as well extract it early instead of late.

+  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 +67,7 @@ Future<Uint8List> collectBytes(Stream<List<int>> source) {
         completer.complete(_collect(length, byteLists));
       },
       cancelOnError: true);
-  return completer.future;
+  return result(subscription, completer);

[nweiz, 2017/01/27 21:56:00]

I really wish we had efficient multiple returns -_-.

[Lasse Reichstein Nielsen, 2017/01/30 15:06:33]

Not sure what "multiple returns" means here, but I'm sure I wish for it in any
case :)

[nweiz, 2017/01/30 22:21:26]

So we could write "return subscription, completer;" here and "var subscription,
completer = _collectBytes(source);" in callsites.

[Lasse Reichstein Nielsen, 2017/01/31 11:53:54]

Oh, I absolutely would love to have that.
It's so annoying that you can call a method with two arguments, but not get two
results back.
(As a continuation-passing-style aficionado, I see no difference between the two
:).

 }
 
 // Join a lists of bytes with a known total length into a single [Uint8List].