Workers

A worker reserves messages from the broker, runs the matching task handlers, stores results, schedules retries and acks. Concurrency is cooperative async inside one isolate.

import 'package:gisila_queue/gisila_queue.dart';

final worker = app.worker(queues: ['default', 'emails'], concurrency: 8);
await worker.run(); // returns after worker.stop()

Worker

class Worker {
  Worker(
    GisilaQueue app, {
    required List<String> queues,
    int concurrency = 1,
    Duration reserveTimeout = const Duration(seconds: 5),
    String? id,
    Duration heartbeatInterval = const Duration(seconds: 10),
    String? deadLetterQueue = Worker.defaultDeadLetterQueue,
    int maxRequeues = 10,
    int maxRedeliveries = 5,
    Duration unknownTaskBackoff = const Duration(seconds: 1),
  });

  static const defaultDeadLetterQueue = 'dead_letter';

  TaskObserver? onSuccess;
  TaskObserver? onFailure;
  TaskObserver? onRetry;

  bool get isRunning;
  Future<void> run();
  Future<void> stop({Duration? timeout});
}

When to use: in the worker process. Build it with app.worker(), or construct it directly to set heartbeatInterval. For a ready-made entrypoint with flags and signal handling, use WorkerCli.

Parameter / member	Type	Default	Description
app	GisilaQueue	required	The app with the registered tasks.
queues	List<String>	required (app.worker defaults to [app.defaultQueue])	Queues to consume, in priority order: earlier queues are drained first. Don’t include the dead-letter queue.
concurrency	int	1	Messages processed at once in this process (one consume loop each).
reserveTimeout	Duration	5 s	How long each reserve waits before looping. Split into 500 ms slices so stop() is prompt.
id	String?	'<hostname>:<pid>'	Identity in events and logs.
heartbeatInterval	Duration	10 s	How often a workerHeartbeat event is emitted.
deadLetterQueue	String?	'dead_letter'	Where messages that failed for good go. null turns dead-lettering off.
maxRequeues	int	10	How often a message may come back via RejectTask(requeue: true) or because its task name is unknown, before it is dead-lettered.
maxRedeliveries	int	5	How often the broker may redeliver an unacked message (its worker kept dying) before it is dead-lettered instead of run.
unknownTaskBackoff	Duration	1 s	First delay before a message with an unknown task name is offered again; doubles per requeue, capped at 1 minute.
onSuccess / onFailure / onRetry	TaskObserver?	null	In-process hooks; see below.
isRunning	bool		Whether run() is active.

run()

Starts concurrency consume loops, subscribes to the control channel, emits workerOnline, starts heartbeats, and completes only after stop(). Calling it twice is a no-op. A worker on the default namespace gq logs a warning.

stop()

Future<void> stop({Duration? timeout})

Stops taking new messages and waits for running tasks to finish. With timeout, it stops waiting after that long; tasks still running stay unacked, so a broker with a visibility timeout (Redis) redelivers them later. Emits workerOffline.

Processing order

For each reserved message the worker:

1. emits taskReceived;
2. if the task name is unknown: requeues it with a growing delay (up to maxRequeues), then records failure and dead-letters it;
3. if redeliveries > maxRedeliveries: records failure and dead-letters it;
4. if revoked (live broadcast, or revoked in the backend): records revoked;
5. if past expires: records revoked;
6. if over the task’s rateLimit: re-enqueues with an ETA, records scheduled;
7. stores started, runs the handler (with timeLimit if set), then records success, retries, rejects or fails as described in Retries and acks;
8. acks the message.

While a message is processed on a broker with a visibility timeout, the worker calls touch every third of the timeout (at least every 50 ms) so slow tasks aren’t redelivered.

An error outside the handler (backend down, ack failed) is logged, the slot sleeps one second and continues. The message stays unacked and a reliable broker redelivers it.

TaskObserver

typedef TaskObserver = void Function(TaskMessage message, StoredResult result);

When to use: quick in-process metrics or logging for one worker, without setting up an event sink. For cross-process monitoring use events.

Hook	Fires
onSuccess	After a successful run is recorded.
onFailure	After a terminal failure (handler error with no retry, retries exhausted, too many requeues). Not for unknown-task or redelivery-limit failures.
onRetry	After a retry is enqueued, with the next message and a retry result.

final worker = app.worker(concurrency: 4)
  ..onFailure = (msg, result) => alert('${msg.task} ${msg.id}: ${result.error}');

See also: Run workers, Concurrency, CLI.