SimpleObserver & Debouncer demo

main.dart

import 'dart:async';
import 'dart:math';

void main() {
  /// Use of the SimpleObservable base class.
  final observable = SimpleObservable<String>(printCallback);
  observable.values.listen(printStream);

  /// Recursively listens to [nextValue] and prints changes.
  printFuture(observable);

  observable.value = 'a';
  observable.value = 'b';
  observable.value = 'c';

  /// Use of the Debouncer class.

  final debouncer =
      Debouncer<String>(Duration(milliseconds: 250), printCallback);
  debouncer.values.listen(printStream);
  printFuture(debouncer);

  /// Change the value multiple times before the debounce timer runs out.
  debouncer.value = '';
  final timer = Timer.periodic(Duration(milliseconds: 200), (_) {
    debouncer.value += 'x';
  });

  Future.delayed(Duration(milliseconds: 1000)).then((_) async {
    /// Cancels the above timer.
    timer.cancel();

    /// Make another change after the debouncer emits its value.
    await Future.delayed(Duration(milliseconds: 500));
    debouncer.value = 'hi';
  });

  // Multiple listeners are supported.
  debouncer.values.listen((value) => print('Stream2: $value'));

  // Transforming the stream.
  debouncer.values
      .transform<List<String>>(StreamTransformer.fromHandlers(
          handleData: (value, sink) => sink.add(['Transformed', value])))
      .listen(print);

  funWithRandom();
}

void printCallback(String value) => print('Callback: $value');
void printStream(String value) => print('Stream: $value');
void printFuture(SimpleObservable obs) => obs.nextValue.then((value) {
      print('Future: $value');
      printFuture(obs);
    });

void funWithRandom() async {
  final d = Debouncer<int>(Duration(milliseconds: 250));
  final rng = Random();
  int i = 0;
  int count = 0;
  d.values.listen((value) {
    double fraction = ((++count / value) * 1000).round() / 1000;
    print('count: $count, current: $value, pct: $fraction');
    if ((count >= 25 && (fraction > 0.27 || fraction < 0.23)) || count >= 100)
      d.cancel();
  });
  while (!d.canceled) {
    await Future.delayed(Duration(milliseconds: 100 + rng.nextInt(200)));
    d.value = ++i;
  }
  print('Done.');
}

// Output:
//
// Callback: a
// Future: a
// Stream: a
// Callback: b
// Future: b
// Stream: b
// Callback: c
// Future: c
// Stream: c
// Callback: xxxxx
// Future: xxxxx
// Stream: xxxxx
// Stream2: xxxxx
// [Transformed, xxxxx]
// Callback: hi
// Future: hi
// Stream: hi
// Stream2: hi
// [Transformed, hi]
//
// Random output...

// =======================================================================

/// A simple class that allows being notified of changes [value] via the
/// [onValue] callback, the [nextValue] Future, or the [values] Stream.
///
/// Any combination of [onValue], [nextValue], and [values] can be used to
/// listen for changes.
///
/// Once canceled, it cannot be reused. Instead, create another instance.
class SimpleObservable<T> {
  SimpleObservable([this.onValue]);

  final void Function(T value) onValue;

  var _completer = Completer<T>();

  bool _canceled = false;
  bool get canceled => _canceled;

  T _value;

  /// The current value of this observable.
  T get value => _value;
  set value(T val) {
    if (!canceled) {
      _value = val;
      // Delaying notify() allows the Future and Stream to update correctly.
      Future.delayed(Duration(microseconds: 1), () => _notify(val));
    }
  }

  /// Alias for [value] setter. Good for passing to a Future or Stream.
  void setValue(T val) => value = val;

  void _notify(T val) {
    if (onValue != null) onValue(val);
    // Completing with a microtask allows a new completer to be constructed
    // before listeners of [nextValue] are called, allowing them to listen to
    // nextValue again if desired.
    _completer.complete(Future.microtask(() => val));
    _completer = Completer<T>();
  }

  Future<T> get nextValue => _completer.future;
  Stream<T> get values async* {
    while (!canceled) yield await nextValue;
  }

  /// Permanently disables this observable. Further changes to [value] will be
  /// ignored, the outputs [onValue], [nextValue], and [values] will not be
  /// called again.
  void cancel() => _canceled = true;
}

/// Debounces value changes by updating [onValue], [nextValue], and [values]
/// only after [duration] has elapsed without additional changes.
class Debouncer<T> extends SimpleObservable<T> {
  Debouncer(this.duration, [void Function(T value) onValue]) : super(onValue);
  final Duration duration;
  Timer _timer;

  /// The most recent value, without waiting for the debounce timer to expire.
  @override
  T get value => super.value;

  set value(T val) {
    if (!canceled) {
      _value = val;
      _timer?.cancel();
      _timer = Timer(duration, () {
        if (!canceled) {
          _notify(value);
        }
      });
    }
  }

  @override
  void cancel() {
    super.cancel();
    _timer?.cancel();
  }
}