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
|
+ return _collectBytes<Future<Uint8List>>(source, _futureResult); |
nweiz
2017/01/27 21:56:00
I think you shouldn't need an explicit generic typ
Lasse Reichstein Nielsen
2017/01/30 15:06:33
It probably doesn't even need to be generic, I can
nweiz
2017/01/30 22:21:26
It does need to be generic in strong mode—but I be
Lasse Reichstein Nielsen
2017/01/31 11:53:54
I've removed all the explicit type declarations (e
|
+} |
+ |
+/// 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, an
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 on
Lasse Reichstein Nielsen
2017/01/30 15:06:33
True. Might as well extract it early instead of la
|
+ 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
nweiz
2017/01/30 22:21:26
So we could write "return subscription, completer;
Lasse Reichstein Nielsen
2017/01/31 11:53:54
Oh, I absolutely would love to have that.
It's so
|
} |
// Join a lists of bytes with a known total length into a single [Uint8List]. |