Throttle improvements

Author: simolus3

Diff for: packages/sqlite_async/lib/src/update_notification.dart

@@ -62,102 +62,63 @@ class UpdateNotification {
   static StreamTransformer<UpdateNotification, UpdateNotification>
       filterTablesTransformer(Iterable<String> tables) {
     Set<String> normalized = {for (var table in tables) table.toLowerCase()};
-
-    return StreamTransformer.fromBind(
-        (source) => source.where((data) => data.containsAny(normalized)));
+    return StreamTransformer<UpdateNotification,
+        UpdateNotification>.fromHandlers(handleData: (data, sink) {
+      if (data.containsAny(normalized)) {
+        sink.add(data);
+      }
+    });
   }
 }
 
-/// Given an [input] stream, returns a stream that will throttle events for each
-/// listener.
+/// Given a broadcast stream, return a singular throttled stream that is throttled.
+/// This immediately starts listening.
 ///
 /// Behaviour:
 ///   If there was no event in "timeout", and one comes in, it is pushed immediately.
 ///   Otherwise, we wait until the timeout is over.
 Stream<T> _throttleStream<T extends Object>(Stream<T> input, Duration timeout,
-    {bool throttleFirst = false, T Function(T, T)? add, T? addOne}) {
-  return Stream.multi(
-    (downstream) {
-      Timer? activeDelay;
-      T? pendingData;
-
-      bool needsToDelay() {
-        return downstream.isPaused || activeDelay != null;
-      }
-
-      void scheduleDelay() {
-        assert(activeDelay == null);
-
-        if (!needsToDelay()) {
-          activeDelay = Timer(timeout, () {
-            activeDelay = null;
-
-            if (!needsToDelay()) {
-              if (pendingData case final pending?) {
-                pendingData = null;
-                downstream.addSync(pending);
-                scheduleDelay();
-              }
-            }
-          });
-        }
-      }
-
-      void cancelTimer() {
-        activeDelay?.cancel();
-        activeDelay = null;
-      }
-
-      var listener = input.listen(
-        (data) {
-          if (needsToDelay()) {
-            // We can't send this yet, so combine / replace the value to send when
-            // the wait is over.
-            if (pendingData != null && add != null) {
-              pendingData = add(pendingData as T, data);
-            } else {
-              pendingData = data;
-            }
-          } else {
-            // We can forward this event directly, but need to wait for a timeout
-            // before sending the next event.
-            assert(pendingData == null);
-            downstream.addSync(data);
-            scheduleDelay();
-          }
-        },
-        onError: downstream.addErrorSync,
-        onDone: () {
-          cancelTimer();
-          if (pendingData case final pending?) {
-            downstream.addSync(pending);
-          }
-          downstream.closeSync();
-        },
-      );
-
-      if (addOne != null) {
-        downstream.add(addOne);
-      }
-      if (throttleFirst) {
-        scheduleDelay();
-      }
+    {bool throttleFirst = false, T Function(T, T)? add, T? addOne}) async* {
+  var nextPing = Completer<void>();
+  var done = false;
+  T? lastData;
+
+  var listener = input.listen((data) {
+    if (lastData != null && add != null) {
+      lastData = add(lastData!, data);
+    } else {
+      lastData = data;
+    }
+    if (!nextPing.isCompleted) {
+      nextPing.complete();
+    }
+  }, onDone: () => done = true);
 
-      downstream.onResume = () {
-        if (!needsToDelay()) {
-          if (pendingData case final pending?) {
-            pendingData = null;
-            downstream.add(pending);
-            scheduleDelay();
-          }
-        }
-      };
+  try {
+    if (addOne != null) {
+      yield addOne;
+    }
+    if (throttleFirst) {
+      await Future.delayed(timeout);
+    }
+    while (!done) {
+      // If a value is available now, we'll use it immediately.
+      // If not, this waits for it.
+      await nextPing.future;
+      // Capture any new values coming in while we wait.
+      nextPing = Completer<void>();
+      T data = lastData as T;
+      // Clear before we yield, so that we capture new changes while yielding
+      lastData = null;
+      yield data;
+      // Wait a minimum of this duration between tasks
+      await Future.delayed(timeout);
+    }
+  } finally {
+    if (lastData case final data?) {
+      yield data;
+    }
 
-      downstream.onCancel = () {
-        cancelTimer();
-        return listener.cancel();
-      };
-    },
-    isBroadcast: input.isBroadcast,
-  );
+    await listener.cancel();
+  }
 }

Diff for: packages/sqlite_async/test/update_notification_test.dart

@@ -81,7 +81,7 @@ void main() {
           control.elapse(halfTimeout);
 
           sub.cancel();
-          control.flushMicrotasks();
+          control.flushTimers();
 
           expect(cancelled, isTrue);
           expect(control.pendingTimers, isEmpty);
@@ -104,7 +104,7 @@ void main() {
             ..add(UpdateNotification({'b'}))
             ..close();
 
-          control.flushMicrotasks();
+          control.flushTimers();
           expect(events, [
             UpdateNotification({'a', 'b'})
           ]);