debouncer.dart

import 'dart:async';

import 'package:clock/clock.dart';

/// Debounces an asynchronous operation, calling [onUpdate] after [add] is used
/// to update [value].
///
/// This class makes a few useful guarantees:
///
/// 1. [onUpdate] will not be called more frequent than [minDuration].
/// 2. [onUpdate] will be called at least every [maxDuration] if [add] is called
///    more frequent than [minDuration].
/// 3. [onUpdate] will not be called concurrently, e.g. if [add] is called while
///    an asynchronous [onUpdate] is in progress, the debouncer will wait until
///    it finishes.
class Debouncer<T> {
  Debouncer({
    required this.minDuration,
    this.maxDuration,
    required this.onUpdate,
  });

  final Duration minDuration;
  final Duration? maxDuration;
  final FutureOr<void> Function(T value) onUpdate;

  late T _value;
  T get value => _value;

  Timer? _timer;
  var _isUpdating = false;
  var _shouldUpdate = false;
  DateTime? _lastUpdate;

  void _update({DateTime? now}) async {
    _timer?.cancel();
    _timer = null;
    if (_isUpdating) {
      _shouldUpdate = true;
      return;
    }
    _lastUpdate = now ?? clock.now();
    _isUpdating = true;
    try {
      await onUpdate(_value);
    } catch (error, stackTrace) {
      Zone.current.handleUncaughtError(error, stackTrace);
    }
    _isUpdating = false;
    if (_shouldUpdate) {
      _shouldUpdate = false;
      _update();
    }
  }

  void add(T value) {
    _value = value;

    _timer?.cancel();
    _timer = null;

    final now = clock.now();
    _lastUpdate ??= now;

    var duration = minDuration;
    if (maxDuration != null) {
      final newDuration = _lastUpdate!.add(maxDuration!).difference(now);
      if (newDuration < duration) {
        duration = newDuration;
      }
    }

    if (duration.isNegative || duration == Duration.zero) {
      _update(now: now);
    } else {
      _timer = Timer(duration, _update);
    }
  }
}

debouncer_test.dart

import 'dart:async';

import 'package:clock/clock.dart';
import 'package:dub_app/core/common/debouncer.dart';
import 'package:fake_async/fake_async.dart';
import 'package:flutter_test/flutter_test.dart';

void main() {
  test('debouncing', () {
    // Inspired by https://rxmarbles.com/ - a way to visualize how streams
    // process events.
    void marbles(
      String input,
      String output,
    ) {
      final startTime = DateTime(2022, 7, 11);
      final async = FakeAsync(initialTime: startTime);

      final outputs = <String>[];
      final waits = <int>[];
      final timing = <int>[];
      for (var i = 0; i < input.length; i++) {
        final char = output[i];
        if (char.trim().isEmpty) continue;
        if (outputs.isNotEmpty && char == outputs.last) {
          waits[waits.length - 1]++;
        } else {
          timing.add(i);
          outputs.add(char);
          waits.add(1);
        }
      }
      var updating = false;
      var index = 0;

      final debouncer = Debouncer<String>(
        minDuration: const Duration(seconds: 5),
        maxDuration: const Duration(seconds: 10),
        onUpdate: (value) async {
          final now = clock.now().difference(startTime).inSeconds;
          expect(updating, isFalse, reason: 'Concurrent update at $now');
          expect(
            outputs,
            hasLength(greaterThan(index)),
            reason: 'unexpected $value at $now',
          );
          expect(value, outputs[index], reason: 'Element #$index at $now');
          expect(now, timing[index], reason: 'Timing for $value (#$index)');
          updating = true;
          await Future<void>.delayed(Duration(seconds: 1 * waits[index]));
          index++;
          updating = false;
        },
      );

      var didError = false;
      final testZone = Zone.current;

      for (final char in input.split('')) {
        if (char.trim().isNotEmpty) {
          async.run<void>((e) {
            runZonedGuarded(
              () => debouncer.add(char),
              (error, stackTrace) {
                didError = true;
                testZone.handleUncaughtError(error, stackTrace);
              },
            );
          });
        }
        async.elapse(const Duration(seconds: 1));
        if (didError) {
          throw TestFailure('Error occurred');
        }
      }
      async.flushTimers();

      expect(index, outputs.length, reason: 'Incomplete outputs');
    }

    // minDuration is 5 seconds, maxDuration is 10 seconds.

    // Adding elements every 5 seconds produces a slightly delayed stream.
    marbles(
      'a    b    c     ',
      '     a    b    c',
    );
    marbles(
      'a         b     ',
      '     a         b',
    );

    // Adding elements quicker than 5 seconds causes it to wait another 5
    // seconds before updating.
    marbles(
      'a   b          ',
      '         b     ',
    );

    // Constantly adding elements quicker than 5 causes it to update every 10
    // seconds (maxDuration), or 5 seconds after it stops.
    marbles(
      'a   b   c   d   e    ',
      '          c         e',
    );

    // Updating might take a little while, e.g. if we're doing a network
    // request represented by the long 'aaaaa'. The long update doesn't affect
    // timing when there's no overlap.
    marbles(
      'a      b     ',
      '     aaaaa  b',
    );

    // Updating happens again immediately after a previous update blocked it.
    marbles(
      'a    b        ',
      '     aaaaaaaab',
    );

    // Updates always use the latest value.
    marbles(
      'a    b    c   ',
      '     aaaaaaaac',
    );
    marbles(
      'a    b    c    d    e ',
      '     aaaaaaaacccccccce',
    );
  });
}