Fix handling pause immediately after emit

Author: simolus3

packages/sqlite_async/lib/src/update_notification.dart

@@ -97,6 +97,7 @@ Stream<T> _throttleStream<T extends Object>({
   return Stream.multi((listener) {
     T? pendingData;
     Timer? activeTimeoutWindow;
+    var needsTimeoutWindowAfterResume = false;
 
     /// Add pending data, bypassing the active timeout window.
     ///
@@ -106,6 +107,7 @@ Stream<T> _throttleStream<T extends Object>({
         pendingData = null;
         listener.addSync(data);
         activeTimeoutWindow?.cancel();
+        activeTimeoutWindow = null;
         return true;
       } else {
         return false;
@@ -118,10 +120,14 @@ Stream<T> _throttleStream<T extends Object>({
       if (activeTimeoutWindow == null && !listener.isPaused) {
         final didAdd = addPendingEvents();
         if (didAdd) {
-          activeTimeoutWindow = Timer(timeout, () {
-            activeTimeoutWindow = null;
-            maybeEmit();
-          });
+          if (listener.isPaused) {
+            needsTimeoutWindowAfterResume = true;
+          } else {
+            activeTimeoutWindow = Timer(timeout, () {
+              activeTimeoutWindow = null;
+              maybeEmit();
+            });
+          }
         }
       }
     }
@@ -152,11 +158,11 @@ Stream<T> _throttleStream<T extends Object>({
     }
 
     final subscription = input.listen(onData, onError: onError, onDone: onDone);
-    var needsTimeoutWindowAfterResume = false;
 
     listener.onPause = () {
       needsTimeoutWindowAfterResume = activeTimeoutWindow != null;
       activeTimeoutWindow?.cancel();
+      activeTimeoutWindow = null;
     };
     listener.onResume = () {
       if (needsTimeoutWindowAfterResume) {

packages/sqlite_async/test/update_notification_test.dart

@@ -47,6 +47,61 @@ void main() {
         });
       });
 
+      test('increases delay after pause', () {
+        fakeAsync((control) {
+          final source = StreamController<UpdateNotification>(sync: true);
+          final events = <UpdateNotification>[];
+
+          final sub = UpdateNotification.throttleStream(source.stream, timeout)
+              .listen(null);
+          sub.onData((event) {
+            events.add(event);
+            sub.pause();
+          });
+
+          source.add(UpdateNotification({'a'}));
+          control.elapse(timeout);
+          expect(events, hasLength(1));
+
+          // Assume the stream stays paused for the timeout window that would
+          // be created after emitting the notification.
+          control.elapse(timeout * 2);
+          source.add(UpdateNotification({'b'}));
+          control.elapse(timeout * 2);
+
+          // A full timeout needs to pass after resuming before a new item is
+          // emitted.
+          sub.resume();
+          expect(events, hasLength(1));
+
+          control.elapse(halfTimeout);
+          expect(events, hasLength(1));
+          control.elapse(halfTimeout);
+          expect(events, hasLength(2));
+        });
+      });
+
+      test('does not introduce artificial delay in pause', () {
+        fakeAsync((control) {
+          final source = StreamController<UpdateNotification>(sync: true);
+          final events = <UpdateNotification>[];
+
+          final sub = UpdateNotification.throttleStream(source.stream, timeout)
+              .listen(events.add);
+
+          // Await the initial delay
+          control.elapse(timeout);
+
+          sub.pause();
+          source.add(UpdateNotification({'a'}));
+          // Resuming should not introduce a timeout window because no window
+          // was active when the stream was paused.
+          sub.resume();
+          control.flushMicrotasks();
+          expect(events, hasLength(1));
+        });
+      });
+
       test('merges events', () {
         fakeAsync((control) {
           final source = StreamController<UpdateNotification>(sync: true);