GisilaQueue class

The central application object, analogous to a Celery Celery() app.

It owns the task registry, the Broker, and the optional ResultBackend, and is the single thing both producers and workers construct. Register tasks with task; enqueue them through the returned TaskRef; process them by building a worker.

final app = GisilaQueue(
  broker: RedisBroker(host: 'localhost'),
  backend: RedisResultBackend(host: 'localhost'),
);

final add = app.task<int>('math.add', (ctx) async =>
    (ctx.arg(0) as int) + (ctx.arg(1) as int));

// producer
final r = await add.delay([2, 3]);
print(await r.get()); // 5  (once a worker has run)

// worker process
await app.worker().run();

Apps that share a Redis database or RabbitMQ vhost need their own namespace (every component's prefix); GisilaQueue.redis sets it once for all of them. The constructor throws ArgumentError when the components passed to it use different prefixes.

Constructors

GisilaQueue({required Broker broker, ResultBackend? backend, String defaultQueue = 'default', QueueEventSink? events, QueueLogger? logger, Uuid? idGenerator, ControlChannel? controlChannel, Router? router, BeatLock? beatLock, BeatStateStore? beatStateStore})
GisilaQueue.redis({required String namespace, String host = 'localhost', int port = 6379, RedisOptions options = const RedisOptions(), bool results = true, bool control = true, bool events = false, Duration visibilityTimeout = const Duration(minutes: 5), Duration resultTtl = const Duration(hours: 24), String defaultQueue = 'default', QueueLogger? logger, Uuid? idGenerator, Router? router})
A Redis-backed app whose broker, result backend, control channel, event sink (with events), beat lock and beat state store all use namespace as their prefix. Give each app on a shared Redis its own namespace; workers, producers and beat of one app must agree on it.
factory

Properties

backend → ResultBackend?
Optional store for task states / return values. When null, tasks run fire-and-forget and AsyncResult.get is unavailable.
final
beatLock → BeatLock?
Lock a Beat built for this app uses when given none. Set by GisilaQueue.redis.
final
beatStateStore → BeatStateStore?
State store a Beat built for this app uses when given none. Set by GisilaQueue.redis.
final
broker → Broker
Transport used to move messages between producers and workers.
final
controlChannel → ControlChannel
Broadcast channel for AsyncResult.revoke and Worker queue pause/resume. Defaults to NoopControlChannel (zero overhead); supply an InMemoryControlChannel or RedisControlChannel to enable it — see ControlChannel.
final
defaultQueue → String
Queue used when a task / call specifies none and router (if any) doesn't match.
final
events → QueueEventSink
Observability sink fed lifecycle QueueEvents by producers and workers. Defaults to NoopEventSink (zero overhead); supply an InMemoryEventBus or RedisEventSink to drive a GisilaRose dashboard.
final
hashCode → int
The hash code for this object.
no setterinherited
logger → QueueLogger
final
namespace → String?
The prefix shared by this app's namespaced components (Redis or RabbitMQ), or null when none is namespaced (e.g. all in-memory).
final
registry → Map<String, Task>
Registered tasks, keyed by name (read-only view).
no setter
router → Router?
Optional declarative routing consulted by send whenever a call doesn't pin an explicit queue — see Router.
final
runtimeType → Type
A representation of the runtime type of the object.
no setterinherited

Methods

chain<T>(List<Signature> signatures) → Chain<T>
Builds a sequential pipeline of signatures — see Chain. Call Chain.apply to enqueue it; the returned AsyncResult resolves once the last step finishes.
chord<T>(List<Signature> signatures, Signature callback) → Chord<T>
Builds a Group of signatures plus a callback that fires once every member has finished — see Chord. Call Chord.apply to enqueue it; the returned AsyncResult resolves once the callback itself finishes, with the member results passed as its first argument.
close() → Future<void>
Closes the broker, backend, and event-sink connections.
group<T>(List<Signature> signatures) → Group<T>
Builds a parallel set of signatures — see Group. Call Group.apply to enqueue every member at once and get back a GroupResult handle for awaiting all of them.
lookup(String name) → Task?
Looks up a registered task by name, or null.
newTaskId() → String
Generates a fresh unique id — e.g. so chain can pre-assign every step's id before enqueuing any of them.
noSuchMethod(Invocation invocation) → dynamic
Invoked when a nonexistent method or property is accessed.
inherited
replayDeadLetters({String queue = Worker.defaultDeadLetterQueue, int limit = 100}) → Future<int>
Moves up to limit messages from the dead-letter queue back to the queue each one failed on, with its retry, requeue and redelivery counters reset, and returns how many were moved. Fix whatever made them fail first, or they will end up back on queue.
resultFor<T>(String id) → AsyncResult<T>
Returns an AsyncResult handle for a known task id (e.g. one persisted elsewhere) without enqueuing anything.
send<T>(String taskName, {List<Object?> args = const [], Map<String, Object?> kwargs = const {}, String? queue, DateTime? eta, int maxRetries = 0, DateTime? expires, String? id, List<Map<String, Object?>>? chain, Map<String, Object?>? chord, String? dedupeKey, Duration dedupeFor = const Duration(hours: 1)}) → Future<AsyncResult<T>>
Low-level enqueue used by TaskRef. Produces a TaskMessage, records a pending (or scheduled) state in the backend, and hands it to the broker.
task<T>(String name, TaskHandler handler, {String? queue, int maxRetries = 0, Duration retryBackoff = const Duration(seconds: 1), Duration retryBackoffMax = const Duration(minutes: 10), bool retryJitter = true, bool autoRetry = false, Duration? timeLimit, Duration? softTimeLimit, RateLimit? rateLimit}) → TaskRef<T>
Registers a task handler under name and returns a typed TaskRef for enqueuing it. T is the handler's result type, surfaced through AsyncResult.
toString() → String
A string representation of this object.
inherited
worker({List<String>? queues, int concurrency = 1, Duration reserveTimeout = const Duration(seconds: 5), String? id, String? deadLetterQueue = Worker.defaultDeadLetterQueue, int maxRequeues = 10, int maxRedeliveries = 5, Duration unknownTaskBackoff = const Duration(seconds: 1)}) → Worker
Builds a Worker bound to this app.

Operators

operator ==(Object other) → bool
The equality operator.
inherited