buffer values when paused

Author: hoc081098

lib/src/common.dart

@@ -8,10 +8,10 @@ Stream<R> toStreamWithTransform<T extends Listenable, R>(
   T listenable,
   R Function(T) transform,
 ) {
-  late StreamController<R> controller;
+  final controller = StreamController<R>();
   VoidCallback? listener;
 
-  final onListenOrOnResume = () {
+  controller.onListen = () {
     assert(listener == null);
     try {
       listenable
@@ -21,26 +21,15 @@ Stream<R> toStreamWithTransform<T extends Listenable, R>(
     }
   };
 
-  final createOnPauseOrOnCancel = ([bool closeOnError = false]) {
+  controller.onCancel = () {
     return () {
       assert(listener != null);
       try {
         listenable.removeListener(listener!);
         listener = null;
-      } catch (_ /*Ignore*/) {
-        if (identical(closeOnError, true)) {
-          controller.close();
-        }
-      }
+      } catch (_ /*Ignore*/) {}
     };
   };
 
-  controller = StreamController<R>(
-    onListen: onListenOrOnResume,
-    onPause: createOnPauseOrOnCancel(true),
-    onResume: onListenOrOnResume,
-    onCancel: createOnPauseOrOnCancel(),
-  );
-
   return controller.stream;
 }

test/listenable_stream_test.dart

@@ -69,11 +69,11 @@ void main() {
       final subscription = stream.listen(
         expectAsync1(
           (v) => expect(v, changeNotifier),
-          count: 1,
+          count: 4,
         ),
       )..pause();
 
-      // no effect
+      // buffer
       changeNotifier.notifyListeners();
       changeNotifier.notifyListeners();
       changeNotifier.notifyListeners();
@@ -181,30 +181,32 @@ void main() {
       test('not replay', () async {
         final valueNotifier = ValueNotifier(0);
         final stream = valueNotifier.toValueStream();
-        final expected = 4;
+        final expected = [1, 2, 3, 4, 5];
 
+        var i = 0;
         final subscription = stream.listen(
           expectAsync1(
-            (v) => expect(v, expected),
-            count: 1,
+            (v) => expect(v, expected[i++]),
+            count: expected.length,
           ),
         )..pause();
 
-        // no effect
+        // buffer
         valueNotifier.value = 1;
         valueNotifier.value = 2;
         valueNotifier.value = 3;
 
         await Future<void>.delayed(const Duration(milliseconds: 50));
         subscription.resume();
 
-        valueNotifier.value = expected;
+        valueNotifier.value = 4;
+        valueNotifier.value = 5;
       });
 
       test('replay + pause immediately', () async {
         final valueNotifier = ValueNotifier(0);
         final stream = valueNotifier.toValueStream(replayValue: true);
-        final expected = [0, 4, 5];
+        final expected = [0, 1, 2, 3, 4, 5];
 
         var i = 0;
         final subscription = stream.listen(
@@ -215,22 +217,21 @@ void main() {
           ),
         )..pause();
 
-        // no effect
+        // buffer
         valueNotifier.value = 1;
         valueNotifier.value = 2;
         valueNotifier.value = 3;
 
         subscription.resume();
 
-        await pumpEventQueue();
         valueNotifier.value = 4;
         valueNotifier.value = 5;
       });
 
       test('replay + pause after events queue.', () async {
         final valueNotifier = ValueNotifier(0);
         final stream = valueNotifier.toValueStream(replayValue: true);
-        final expected = [0, 4, 5];
+        final expected = [0, 1, 2, 3, 4, 5];
 
         var i = 0;
         final subscription = stream.listen(
@@ -244,7 +245,7 @@ void main() {
         await pumpEventQueue();
         subscription.pause();
 
-        // no effect
+        // buffer
         valueNotifier.value = 1;
         valueNotifier.value = 2;
         valueNotifier.value = 3;