execute<T> method

Implementation

Future<T> execute<T>(FutureOr<T> Function() function) async {
  // Current [_iteration] to ignore the invoke, if mismatched.
  int iteration = _iteration;

  // Start the [Timer] reducing the [_queue].
  _timer ??= Timer.periodic(per, (_) {
    // Remove unlocked [Mutex]es.
    queue.removeWhere((e) => !e.isLocked);

    if (queue.isEmpty) {
      _timer?.cancel();
      _timer = null;
      return;
    }

    final taken = queue.take(requests);
    for (var m in taken) {
      m.release();
    }
  });

  final Mutex mutex = Mutex();
  queue.add(mutex);

  if (queue.length > requests) {
    await mutex.acquire();
  }

  await mutex.acquire();
  try {
    if (iteration != _iteration) {
      throw OperationCanceledException();
    }

    if (function is T) {
      return function();
    }

    return await function();
  } finally {
    if (mutex.isLocked) {
      mutex.release();
    }
  }
}