stream_regulate.dart

import 'dart:async';

extension<T> on Stream<T> {
  Stream<T> regulate(Duration rateLimit) {
    final controller = StreamController<T>();
    Timer? timer;
    DateTime? lastTime;
    T? lastMessage;
    listen(
      (message) {
        lastMessage = message;
        final now = DateTime.now();
        if (lastTime == null || (!now.isBefore(lastTime!.add(rateLimit)) && timer != null)) {
          // This is either the first message or we are past the rate limit
          lastTime = now;
          controller.add(message);
        } else {
          // Schedule timer to the next event
          timer ??= Timer(
            lastTime!.add(rateLimit).difference(now),
            () {
              controller.add(lastMessage as T);
              lastTime = now;
              timer = null;
            },
          );
        }
      },
      onError: controller.addError,
      onDone: controller.close,
    );
    return controller.stream;
  }
}