OLD | NEW |
(Empty) | |
| 1 // Copyright (c) 2017, the Dart project authors. Please see the AUTHORS file |
| 2 // for details. All rights reserved. Use of this source code is governed by a |
| 3 // BSD-style license that can be found in the LICENSE file. |
| 4 |
| 5 import "dart:async"; |
| 6 import "dart:typed_data"; |
| 7 import "cancelable_operation.dart"; |
| 8 |
| 9 /// Collects an asynchronous sequence of byte lists into a single list of bytes. |
| 10 /// |
| 11 /// If the [source] stream emits an error event, |
| 12 /// the collection fails and the returned future completes with the same error. |
| 13 /// |
| 14 /// If any of the input data are not valid bytes, they will be truncated to |
| 15 /// an eight-bit unsigned value in the resulting list. |
| 16 Future<Uint8List> collectBytes(Stream<List<int>> source) { |
| 17 return _collectBytes(source, (_, result) => result); |
| 18 } |
| 19 |
| 20 /// Collects an asynchronous sequence of byte lists into a single list of bytes. |
| 21 /// |
| 22 /// Returns a [CancelableOperation] that provides the result future and a way |
| 23 /// to cancel the collection early. |
| 24 /// |
| 25 /// If the [source] stream emits an error event, |
| 26 /// the collection fails and the returned future completes with the same error. |
| 27 /// |
| 28 /// If any of the input data are not valid bytes, they will be truncated to |
| 29 /// an eight-bit unsigned value in the resulting list. |
| 30 CancelableOperation<Uint8List> collectBytesCancelable( |
| 31 Stream<List<int>> source) { |
| 32 return _collectBytes( |
| 33 source, |
| 34 (subscription, result) => new CancelableOperation.fromFuture(result, |
| 35 onCancel: subscription.cancel)); |
| 36 } |
| 37 |
| 38 /// Generalization over [collectBytes] and [collectBytesCancelable]. |
| 39 /// |
| 40 /// Performs all the same operations, but the final result is created |
| 41 /// by the [result] function, which has access to the stream subscription |
| 42 /// so it can cancel the operation. |
| 43 T _collectBytes<T>( |
| 44 Stream<List<int>> source, |
| 45 T result( |
| 46 StreamSubscription<List<int>> subscription, Future<Uint8List> result)) { |
| 47 var byteLists = <List<int>>[]; |
| 48 var length = 0; |
| 49 var completer = new Completer<Uint8List>.sync(); |
| 50 var subscription = source.listen( |
| 51 (bytes) { |
| 52 byteLists.add(bytes); |
| 53 length += bytes.length; |
| 54 }, |
| 55 onError: completer.completeError, |
| 56 onDone: () { |
| 57 completer.complete(_collect(length, byteLists)); |
| 58 }, |
| 59 cancelOnError: true); |
| 60 return result(subscription, completer.future); |
| 61 } |
| 62 |
| 63 // Join a lists of bytes with a known total length into a single [Uint8List]. |
| 64 Uint8List _collect(int length, List<List<int>> byteLists) { |
| 65 var result = new Uint8List(length); |
| 66 int i = 0; |
| 67 for (var byteList in byteLists) { |
| 68 var end = i + byteList.length; |
| 69 result.setRange(i, end, byteList); |
| 70 i = end; |
| 71 } |
| 72 return result; |
| 73 } |
OLD | NEW |