Add `byteCollector` stream transformer and `collectBytes` 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";
+
+/// A stream transformer that collects byte list events into a single list.
+///
+/// The transformer accumulates lists of bytes, and when the source stream ends,
+/// all the bytes are emitted as a single [Uint8List] event on the resulting
+/// stream.
+///
+/// If any of the input data are not valid bytes, they will be truncated to
+/// an eight-bit unsigned value in the resulting list.
+///
+/// Any errors events are forwarded directly to the resulting stream.
+const StreamTransformer<List<int>, Uint8List> byteCollector =
+    const StreamTransformer<List<int>, Uint8List>(_collectionTransformer);

[nweiz, 2017/01/25 23:00:40]

I'm not sure I see the use of a transformer here. It's only ever going to
produce a stream with a single value—in what circumstance would I use it rather
than collectBytes()?

[Lasse Reichstein Nielsen, 2017/01/26 09:43:05]

It might be a little too speculative :)
I was thinking you might want it because it allows chaining.
  someStream.map(..).where(..).transform(byteCollector).first.then(...)
instead of.
  collectBytes(someStream.map(..).where(..)).then(...)

On the other hand, a future is likely to be used with `await` so there
won't be nothing after the call. 

I was also considering having a version that collected up to a specific size
only, so you could collect a sequence of byte lists of length 5000 , but that
would be a different API anyway, and not this const value.

So, good point, I'll just remove it again.

+
+/// 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.
+///
+/// Equivalent to `source.transform(byteCollector).first`.
+Future<Uint8List> collectBytes(Stream<List<int>> source) async {
+  List<List<int>> byteLists = [];

[nweiz, 2017/01/25 23:00:40]

"var byteLists = <List<int>>[]"

Also "var" for assignments below.

[Lasse Reichstein Nielsen, 2017/01/26 09:43:05]

Done.
(Even though I was actually deliberately creating a dynamic list in spec mode,
it's probably not important in practice :)

+  int length = 0;
+  await for (var list in source) {

[nweiz, 2017/01/25 23:00:40]

I'm pretty sure "await for" is still very slow on most implementations. It's
fairly likely that this will be used by performance-critical networking
applications, so we should probably use .listen() directly.

[Lasse Reichstein Nielsen, 2017/01/26 09:43:05]

Argh.
On one hand, I'm convinced it will never be made fast unless we have important
code that is slowed down by it.
On the other hand, I don't want to make it the user's problem (any more than it
already is).
So, done with listen.

+    byteLists.add(list);
+    length += list.length;
+  }
+  return _collect(length, byteLists);
+}
+
+// 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;
+}
+
+/// Collects byte lists from a stream into a single byte list stream.
+StreamSubscription<Uint8List> _collectionTransformer(
+    Stream<List<int>> stream, bool cancelOnError) {
+  List<List<int>> byteLists = [];
+  int length = 0;
+  var controller = new StreamController<Uint8List>(sync: true);
+  StreamSubscription sourceSubscription = stream.listen(
+      (List<int> data) {
+        byteLists.add(data);
+        length += data.length;
+      },
+      onError: controller.addError,
+      onDone: () {
+        controller.add(_collect(length, byteLists));
+        controller.close();
+      });
+  controller
+    ..onPause = sourceSubscription.pause
+    ..onResume = sourceSubscription.resume
+    ..onCancel = sourceSubscription.cancel;
+  return controller.stream.listen(null, cancelOnError: cancelOnError);
+}