Make StreamBuffer implement StreamConsumer<List<T>> (#467)

google · Jul 12, 2018 · 81da0ba · 81da0ba
1 parent 6d8d55b
commit 81da0ba
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 12 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,6 +1,11 @@
 #### Master
 
    * BREAKING CHANGE: This version requires Dart SDK 2.0.0-dev.61 or later.
+   * BREAKING CHANGE: StreamBuffer has been changed from implementing
+     `StreamConsumer<T>` to `StreamConsumer<List<T>>`. Users of
+     `StreamBuffer<List<T>>` can simply change declarations to
+     `StreamBuffer<T>`. In cases where the generic type is already not a list
+     type, inputs to the list may need to be wrapped in a list.
 
 #### 0.29.0+1 - 2018-03-29
 

diff --git a/lib/src/async/stream_buffer.dart b/lib/src/async/stream_buffer.dart
@@ -43,12 +43,12 @@ class UnderflowError extends Error {
 ///
 /// Throws [UnderflowError] if [throwOnError] is true. Useful for unexpected
 /// [Socket] disconnects.
-class StreamBuffer<T> implements StreamConsumer<T> {
+class StreamBuffer<T> implements StreamConsumer<List<T>> {
   List<T> _chunks = [];
   int _offset = 0;
   int _counter = 0; // sum(_chunks[*].length) - _offset
   List<_ReaderInWaiting<List<T>>> _readers = [];
-  StreamSubscription<T> _sub;
+  StreamSubscription<List<T>> _sub;
 
   final bool _throwOnError;
 
@@ -129,15 +129,15 @@ class StreamBuffer<T> implements StreamConsumer<T> {
   }
 
   @override
-  Future addStream(Stream<T> stream) {
+  Future addStream(Stream<List<T>> stream) {
     var lastStream = _currentStream == null ? stream : _currentStream;
     if (_sub != null) {
       _sub.cancel();
     }
     _currentStream = stream;
     final streamDone = new Completer<Null>();
     _sub = stream.listen((items) {
-      _chunks.add(items);
+      _chunks.addAll(items);
       _counter += items is List ? items.length : 1;
       if (limited && _counter >= limit) {
         _sub.pause();

diff --git a/test/async/stream_buffer_test.dart b/test/async/stream_buffer_test.dart
@@ -21,7 +21,7 @@ import 'package:quiver/async.dart';
 void main() {
   group("StreamBuffer", () {
     test("returns orderly overlaps", () {
-      StreamBuffer<List<int>> buf = new StreamBuffer();
+      StreamBuffer<int> buf = new StreamBuffer();
       new Stream.fromIterable([
         [1],
         [2, 3, 4],
@@ -37,7 +37,7 @@ void main() {
     }, tags: ['fails-on-dartdevc']);
 
     test("respects pausing of stream", () {
-      StreamBuffer<List<int>> buf = new StreamBuffer()..limit = 2;
+      StreamBuffer<int> buf = new StreamBuffer()..limit = 2;
       new Stream.fromIterable([
         [1],
         [2],
@@ -54,7 +54,7 @@ void main() {
     }, tags: ['fails-on-dartdevc']);
 
     test("throws when reading too much", () {
-      StreamBuffer<List<int>> buf = new StreamBuffer()..limit = 1;
+      StreamBuffer<int> buf = new StreamBuffer()..limit = 1;
       new Stream.fromIterable([
         [1],
         [2]
@@ -70,8 +70,12 @@ void main() {
 
     test("allows patching of streams", () {
       StreamBuffer<int> buf = new StreamBuffer();
-      new Stream.fromIterable([1, 2]).pipe(buf).then((_) {
-        return new Stream.fromIterable([3, 4]).pipe(buf);
+      new Stream.fromIterable([
+        [1, 2]
+      ]).pipe(buf).then((_) {
+        return new Stream.fromIterable([
+          [3, 4]
+        ]).pipe(buf);
       });
       return Future.wait([buf.read(1), buf.read(2), buf.read(1)]).then((vals) {
         expect(vals[0], equals([1]));
@@ -91,14 +95,20 @@ void main() {
         expect(error is UnderflowError, isTrue,
             reason: "!UnderflowError: $error");
       });
-      new Stream.fromIterable([1, 2, 3]).pipe(buf);
+      new Stream.fromIterable([
+        [1, 2, 3]
+      ]).pipe(buf);
       return future;
     });
 
     test("accepts several streams", () async {
       StreamBuffer<int> buf = new StreamBuffer();
-      new Stream.fromIterable([1]).pipe(buf);
-      new Stream.fromIterable([2, 3, 4, 5]).pipe(buf);
+      new Stream.fromIterable([
+        [1]
+      ]).pipe(buf);
+      new Stream.fromIterable([
+        [2, 3, 4, 5]
+      ]).pipe(buf);
       final vals = await buf.read(4);
       expect(vals, equals([2, 3, 4, 5]));
       buf.close();