refactor(pairwise): change return type of pairwise to Stream<List<T>>

CHANGELOG.md

@@ -21,6 +21,10 @@
 
 ### Documentation
 
+### Code refactoring
+
+* Change return type of `pairwise` to `Stream<List<T>>`.
+
 ## 0.27.4 (2022-05-29)
 
 ### Bug fixes

lib/src/transformers/backpressure/buffer.dart

@@ -21,7 +21,7 @@ class BufferStreamTransformer<T>
   /// The [List] is cleared upon every [window] event.
   BufferStreamTransformer(Stream Function(T event) window)
       : super(WindowStrategy.firstEventOnly, window,
-            onWindowEnd: (List<T> queue) => queue, ignoreEmptyWindows: false);
+            onWindowEnd: (queue) => queue, ignoreEmptyWindows: false);
 }
 
 /// Buffers a number of values from the source Stream by count then
@@ -58,9 +58,9 @@ class BufferCountStreamTransformer<T>
   /// the previous one reaches a length of [count].
   BufferCountStreamTransformer(int count, [int startBufferEvery = 0])
       : super(WindowStrategy.onHandler, null,
-            onWindowEnd: (List<T> queue) => queue,
+            onWindowEnd: (queue) => queue,
             startBufferEvery: startBufferEvery,
-            closeWindowWhen: (Iterable<T> queue) => queue.length == count) {
+            closeWindowWhen: (queue) => queue.length == count) {
     if (count < 1) throw ArgumentError.value(count, 'count');
     if (startBufferEvery < 0) {
       throw ArgumentError.value(startBufferEvery, 'startBufferEvery');
@@ -82,8 +82,8 @@ class BufferTestStreamTransformer<T>
   /// emits this [List] whenever the [test] Function yields true.
   BufferTestStreamTransformer(bool Function(T value) test)
       : super(WindowStrategy.onHandler, null,
-            onWindowEnd: (List<T> queue) => queue,
-            closeWindowWhen: (Iterable<T> queue) => test(queue.last));
+            onWindowEnd: (queue) => queue,
+            closeWindowWhen: (queue) => test(queue.last));
 }
 
 /// Extends the Stream class with the ability to buffer events in various ways

lib/src/transformers/backpressure/debounce.dart

@@ -33,7 +33,7 @@ class DebounceStreamTransformer<T> extends BackpressureStreamTransformer<T, T> {
       : super(
           WindowStrategy.everyEvent,
           window,
-          onWindowEnd: (Iterable<T> queue) => queue.last,
+          onWindowEnd: (queue) => queue.last,
           maxLengthQueue: 1,
         );
 }

lib/src/transformers/backpressure/pairwise.dart

@@ -10,13 +10,13 @@ import 'package:rxdart/src/transformers/backpressure/backpressure.dart';
 ///       .pairwise()
 ///       .listen(print); // prints [1, 2], [2, 3], [3, 4]
 class PairwiseStreamTransformer<T>
-    extends BackpressureStreamTransformer<T, Iterable<T>> {
+    extends BackpressureStreamTransformer<T, List<T>> {
   /// Constructs a [StreamTransformer] which buffers events into pairs as a [List].
   PairwiseStreamTransformer()
       : super(WindowStrategy.firstEventOnly, (_) => NeverStream<void>(),
-            onWindowEnd: (Iterable<T> queue) => queue,
+            onWindowEnd: (queue) => queue,
             startBufferEvery: 1,
-            closeWindowWhen: (Iterable<T> queue) => queue.length == 2,
+            closeWindowWhen: (queue) => queue.length == 2,
             dispatchOnClose: false);
 }
 
@@ -31,5 +31,5 @@ extension PairwiseExtension<T> on Stream<T> {
   ///     RangeStream(1, 4)
   ///       .pairwise()
   ///       .listen(print); // prints [1, 2], [2, 3], [3, 4]
-  Stream<Iterable<T>> pairwise() => PairwiseStreamTransformer<T>().bind(this);
+  Stream<List<T>> pairwise() => PairwiseStreamTransformer<T>().bind(this);
 }

lib/src/transformers/backpressure/sample.dart

@@ -19,7 +19,7 @@ class SampleStreamTransformer<T> extends BackpressureStreamTransformer<T, T> {
   /// the sample [Stream].
   SampleStreamTransformer(Stream Function(T event) window)
       : super(WindowStrategy.firstEventOnly, window,
-            onWindowEnd: (Iterable<T> queue) => queue.last);
+            onWindowEnd: (queue) => queue.last);
 }
 
 /// Extends the Stream class with the ability to sample events from the Stream

lib/src/transformers/backpressure/window.dart

@@ -22,7 +22,7 @@ class WindowStreamTransformer<T>
   /// The [Stream] is recreated and starts empty upon every [window] event.
   WindowStreamTransformer(Stream Function(T event) window)
       : super(WindowStrategy.firstEventOnly, window,
-            onWindowEnd: (List<T> queue) => Stream.fromIterable(queue),
+            onWindowEnd: (queue) => Stream.fromIterable(queue),
             ignoreEmptyWindows: false);
 }
 
@@ -61,9 +61,9 @@ class WindowCountStreamTransformer<T>
   /// the previous one reaches a length of [count].
   WindowCountStreamTransformer(int count, [int startBufferEvery = 0])
       : super(WindowStrategy.onHandler, null,
-            onWindowEnd: (List<T> queue) => Stream.fromIterable(queue),
+            onWindowEnd: (queue) => Stream.fromIterable(queue),
             startBufferEvery: startBufferEvery,
-            closeWindowWhen: (Iterable<T> queue) => queue.length == count) {
+            closeWindowWhen: (queue) => queue.length == count) {
     if (count < 1) throw ArgumentError.value(count, 'count');
     if (startBufferEvery < 0) {
       throw ArgumentError.value(startBufferEvery, 'startBufferEvery');
@@ -86,8 +86,8 @@ class WindowTestStreamTransformer<T>
   /// emits this [Stream] whenever the [test] Function yields true.
   WindowTestStreamTransformer(bool Function(T value) test)
       : super(WindowStrategy.onHandler, null,
-            onWindowEnd: (List<T> queue) => Stream.fromIterable(queue),
-            closeWindowWhen: (Iterable<T> queue) => test(queue.last));
+            onWindowEnd: (queue) => Stream.fromIterable(queue),
+            closeWindowWhen: (queue) => test(queue.last));
 }
 
 /// Extends the Stream class with the ability to window

test/transformers/backpressure/pairwise_test.dart

@@ -16,15 +16,38 @@ void main() {
 
     final stream = Rx.range(1, 4).pairwise();
 
-    stream.listen(expectAsync1((result) {
-      // test to see if the combined output matches
-      expect(expectedOutput[count].length, result.length);
-      expect(expectedOutput[count][0], result.elementAt(0));
-      if (expectedOutput[count].length > 1) {
-        expect(expectedOutput[count][1], result.elementAt(1));
-      }
-      count++;
-    }, count: expectedOutput.length));
+    stream.listen(
+      expectAsync1((result) {
+        // test to see if the combined output matches
+        expect(result, expectedOutput[count++]);
+      }, count: expectedOutput.length),
+      onError: expectAsync2((Object e, StackTrace s) {}, count: 0),
+      onDone: expectAsync0(() {}, count: 1),
+    );
+  });
+
+  test('Rx.pairwise.empty', () {
+    expect(Stream<int>.empty().pairwise(), emitsDone);
+  });
+
+  test('Rx.pairwise.single', () {
+    expect(Stream.value(1).pairwise(), emitsDone);
+  });
+
+  test('Rx.pairwise.compatible', () {
+    expect(
+      Stream.fromIterable([1, 2]).pairwise(),
+      isA<Stream<Iterable<int>>>(),
+    );
+
+    Stream<Iterable<int>> s = Stream.fromIterable([1, 2]).pairwise();
+    expect(
+      s,
+      emitsInOrder(<Object>[
+        [1, 2],
+        emitsDone
+      ]),
+    );
   });
 
   test('Rx.pairwise.asBroadcastStream', () async {