Source on GitHub·lib/src/utils/fast_async_queue/

Overview

FastAsyncQueue runs async jobs (AsyncJob) serially in FIFO order—upload queues, sequential APIs, offline sync, etc. Supports manual start() vs auto-start factory, label + JobInfo tracking, in-task retry() with a retry budget, and QueueEvent listeners.

Factory	Behavior	Typical use
FastAsyncQueue()	Jobs wait until await start()	Batch enqueue, run once
FastAsyncQueue.autoStart()	addJob starts processing when idle	Process on arrival, still serial

---

FastAsyncQueue

Internal linked list; at most one job is running. Calling retry() inside catch marks the head job pendingRetry for another run in the same start() loop. retryTime is the allowed retry count (default 1); -1 means unlimited.

Examples

Manual: enqueue then start().

Manual queue

final queue = FastAsyncQueue();

queue.addJob(
  () => Future.delayed(const Duration(seconds: 1), () => uploadChunk(1)),
  label: 'chunk-1',
  description: 'First chunk',
);
queue.addJob(
  () => Future.delayed(const Duration(seconds: 1), () => uploadChunk(2)),
  label: 'chunk-2',
);

await queue.start(); // FIFO

Auto-start

final queue = FastAsyncQueue.autoStart();

queue.addJob(() => syncProfile());  // runs if idle
queue.addJob(() => syncSettings()); // waits if busy

Retry on failure

final queue = FastAsyncQueue.autoStart();

queue.addJob(
  () async {
    try {
      await requestWithTransientError();
    } catch (_) {
      queue.retry(); // within retryTime budget
    }
  },
  label: 'sync-order',
  retryTime: 3,
);

Queue events

Subscribe with addQueueListener for QueueEvent (type, currentQueueSize, jobLabel, timestamp).

final queue = FastAsyncQueue();
queue.addQueueListener((event) {
  debugPrint('$event');
});

Common QueueEventType: queueStart / beforeJob / afterJob / queueEnd, newJobAdded, retryJob / retryLimitReached, queueClosed, queueStopped, violateAddWhenClosed.

API reference

---

Constructors

final manual = FastAsyncQueue();
final autoRun = FastAsyncQueue.autoStart();

	Description
FastAsyncQueue()	Jobs wait for start().
FastAsyncQueue.autoStart()	Calls start() after addJob when not already running.

---

FastAsyncQueue.addJob

Appends a job. After close(), silently rejects and emits violateAddWhenClosed; duplicate label throws DuplicatedLabelException. Default label is an ISO8601 timestamp.

queue.addJob(
  () async => doWork(),
  label: 'job-a',
  description: 'Optional note',
  retryTime: 1,
);

Parameter	Type	Required	Description
job	AsyncJob	yes	Nullary function returning Future.
label	String?	no	Unique id for getJobInfo and events.
description	String?	no	Stored on JobInfo.
retryTime	int	no	Retry budget, default 1; -1 = unlimited.

---

FastAsyncQueue.addJobThrow

Same as addJob, but throws ClosedQueueException when the queue is closed.

queue.addJobThrow(() async => doWork(), label: 'critical');

Parameter	Type	Required	Description
job	AsyncJob	yes	Task to run.
label	String?	no	Same as addJob.
description	String?	no	Same as addJob.
retryTime	int	no	Same as addJob.

---

FastAsyncQueue.start

When non-empty and not already running, drains the queue until empty, stop() interrupts, or the loop clears on stop. No-op when empty, already running, or closed.

await queue.start();

Return	Type	Description
—	Future<void>	Completes when this run finishes (or is stopped).

---

FastAsyncQueue.retry

Call inside the running job (usually in catch). Marks the head job pendingRetry for another run() in this start() loop; exceeds retryTime → failed and retryLimitReached.

try {
  await apiCall();
} catch (_) {
  queue.retry();
}

No parameters.

---

FastAsyncQueue.stop

Hard stop: clears pending list and _map, resets size, emits queueStopped. Optional callBack runs synchronously at the start.

queue.stop(() => onQueueAborted());

Parameter	Type	Required	Description
callBack	Function?	no	Runs when stop begins.

---

FastAsyncQueue.clear

Calls stop(callBack) and clears job metadata.

queue.clear();

Parameter	Type	Required	Description
callBack	Function?	no	Passed to stop.

---

FastAsyncQueue.close

Blocks further addJob (violateAddWhenClosed or ClosedQueueException); queued and running jobs still finish. Emits queueClosed.

queue.close();

No parameters.

---

FastAsyncQueue.addQueueListener

Registers one listener (later registration replaces the previous). Receives all QueueEvents.

queue.addQueueListener((QueueEvent event) { /* ... */ });

Parameter	Type	Required	Description
listener	QueueListener	yes	void Function(QueueEvent event).

---

FastAsyncQueue.getJobInfo / list

final info = queue.getJobInfo('chunk-1');
final all = queue.list();

Method	Return	Description
getJobInfo(label)	JobInfo	Throws InvalidJobLabelException if missing.
list()	List<JobInfo>	Snapshot of all tracked jobs.

JobInfo: label, description, state (JobState), retryCount, maxRetry.

---

Properties

	Type	Description
size	int (getter)	Pending + running count (list length).
isClosed	bool (getter)	Whether close() was called.

---

Types and exceptions

JobState

Value	Meaning
pending	Enqueued, waiting
running	Executing
pendingRetry	Retry requested
done	Success (about to dequeue)
failed	Failed, no more retries (about to dequeue)

Exceptions

Type	When
DuplicatedLabelException	Duplicate label on addJob
ClosedQueueException	addJobThrow on closed queue
InvalidJobLabelException	Unknown label in getJobInfo