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