Add collectBytesCancelable function.

lib/src/byte_collector.dart

 // Copyright (c) 2017, 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.
 
 import "dart:async";
 import "dart:typed_data";
+import "cancelable_operation.dart";
 
 /// Collects an asynchronous sequence of byte lists into a single list of bytes.
 ///
 /// 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.
 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;
       },
       onError: completer.completeError,
       onDone: () {
         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].
 Uint8List _collect(int length, List<List<int>> byteLists) {
   var result = new Uint8List(length);
   int i = 0;
   for (var byteList in byteLists) {
     var end = i + byteList.length;
     result.setRange(i, end, byteList);
     i = end;
   }
   return result;
 }