[FR] Future Runners #16

typedef FutureDescription<T> = Future<T> Function();

abstract class FutureRunner<T> {
  Future<T> call(FutureDescription<T> future);
}

class ExhaustFutureRunner<T> implements FutureRunner<T> {
  Future<T>? _cachedFuture;

  bool get isRunning => _cachedFuture != null;

  Future<T> _manageFuture(Future<T> future) async {
    _cachedFuture = future;
    final value = await future;
    _cachedFuture = null;
    return value;
  }

  @override
  Future<T> call(FutureDescription<T> future) {
    final cache = _cachedFuture;

    return cache != null ? cache : _manageFuture(future());
  }
}

class SequentialFutureRunner<T> implements FutureRunner<T> {
  late final StreamController<FutureDescription<void>> _queue =
      StreamController();
  StreamSubscription<void>? _executionSubscription;

  SequentialFutureRunner();

  Future<void> _startQueue() async {
    _executionSubscription =
        _queue.stream.asyncMap((future) => future()).listen((_) {});
    await Future<void>.delayed(Duration.zero);
  }

  @override
  Future<T> call(FutureDescription<T> future) async {
    if (_executionSubscription == null) await _startQueue();
    final completer = Completer<T>();
    _queue.add(() => future().then(completer.complete));
    return completer.future;
  }

  Future<void> close() async {
    await _executionSubscription?.cancel();
    await _queue.close();
  }
}