GisilaQueue app

GisilaQueue is the one object both producers and workers build. It holds the task registry, the Broker, an optional ResultBackend, and the optional control channel, event sink, router and beat coordination. It is the equivalent of Celery’s Celery() app.

import 'package:gisila_queue/gisila_queue.dart';

GisilaQueue

The application object.

class GisilaQueue {
  GisilaQueue({
    required Broker broker,
    ResultBackend? backend,
    String defaultQueue = 'default',
    QueueEventSink? events,
    QueueLogger? logger,
    Uuid? idGenerator,
    ControlChannel? controlChannel,
    Router? router,
    BeatLock? beatLock,
    BeatStateStore? beatStateStore,
  });

  factory GisilaQueue.redis({required String namespace, ...});
}

When to use: always. Build one per process, register the same tasks in producers and workers, and share it. Use the plain constructor when you want in-memory components, RabbitMQ, or full control over each piece. Use GisilaQueue.redis when everything lives on one Redis.

Parameter	Type	Default	Description
broker	Broker	required	Transport between producers and workers.
backend	ResultBackend?	null	Where states and return values are stored. null means fire-and-forget: AsyncResult.get throws StateError, dedupeKey and chord() are unavailable.
defaultQueue	String	'default'	Queue used when neither the call, the router nor the task names one. Also the default queue of worker().
events	QueueEventSink?	NoopEventSink()	Receives lifecycle events. Needed for GisilaRose.
logger	QueueLogger?	ConsoleQueueLogger()	Logs to stderr at info and above.
idGenerator	Uuid?	const Uuid()	Source of task ids (UUID v4).
controlChannel	ControlChannel?	NoopControlChannel()	Broadcasts revoke and pause/resume. With the default, AsyncResult.revoke only marks the backend.
router	Router?	null	Declarative task-name to queue rules.
beatLock	BeatLock?	null	Used by a Beat built for this app when the beat is given none.
beatStateStore	BeatStateStore?	null	Same, for persisted last-run times.

The constructor checks that every namespaced component (broker, backend, events, control channel, beat lock, beat state store) uses the same prefix and throws ArgumentError naming each one when they differ.

Example

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

GisilaQueue.redis()

Builds an app whose Redis components all share one namespace.

factory 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,
})

When to use: the normal production setup on Redis. It removes the risk of giving components different prefixes. Use the plain constructor for RabbitMQ or mixed setups.

Parameter	Type	Default	Description
namespace	String	required	Prefix for every key and channel. Must be non-empty and contain no whitespace, or ArgumentError is thrown.
host / port	String / int	'localhost' / 6379	Redis address.
options	RedisOptions	RedisOptions()	Password, TLS, database.
results	bool	true	Adds a RedisResultBackend.
control	bool	true	Adds a RedisControlChannel.
events	bool	false	Adds a RedisEventSink (turn on for GisilaRose).
visibilityTimeout	Duration	5 minutes	Passed to RedisBroker.
resultTtl	Duration	24 hours	Passed to RedisResultBackend.
defaultQueue, logger, idGenerator, router			As on the constructor.

It always adds a RedisBeatLock and a RedisBeatStateStore, so a Beat(app) gets leader election and persisted state without more wiring.

Example

final redis = RedisOptions.parseUrl('rediss://app:s3cret@cache:6380/1');
final app = GisilaQueue.redis(
  namespace: 'shop',
  host: redis.host,
  port: redis.port,
  options: redis.options,
  events: true,
);

Properties

Member	Type	Description
broker	Broker	The broker passed in.
backend	ResultBackend?	The backend, or null.
defaultQueue	String	Fallback queue name.
router	Router?	Routing rules.
events	QueueEventSink	Event sink (never null; NoopEventSink by default).
logger	QueueLogger	Logger.
controlChannel	ControlChannel	Control channel (never null).
namespace	String?	The shared prefix of the namespaced components, or null when none is namespaced (all in-memory).
beatLock / beatStateStore		Beat defaults (set by GisilaQueue.redis).
registry	Map<String, Task>	Read-only view of registered tasks.

Methods

task<T>()

Registers a handler under a name and returns a typed TaskRef<T>. Throws ArgumentError if the name is taken. See Tasks for every option.

lookup()

Task? lookup(String name)

Returns the registered Task or null.

send<T>()

Low-level enqueue by task name, used by TaskRef and by code that doesn’t hold a TaskRef. Supports dedupeKey. See Producer API.

resultFor<T>()

AsyncResult<T> resultFor<T>(String id)

Returns an AsyncResult for an id you stored earlier (for example in a database row). Enqueues nothing.

newTaskId()

String newTaskId()

A fresh UUID v4, the same generator used for task ids.

chain<T>(), group<T>(), chord<T>()

Chain<T> chain<T>(List<Signature> signatures)
Group<T> group<T>(List<Signature> signatures)
Chord<T> chord<T>(List<Signature> signatures, Signature callback)

Build workflows. Nothing is sent until you call apply(). See Workflows.

worker()

Worker 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),
})

Builds a Worker bound to this app. queues defaults to [defaultQueue]. To set heartbeatInterval, construct Worker directly.

replayDeadLetters()

Future<int> replayDeadLetters({
  String queue = Worker.defaultDeadLetterQueue,
  int limit = 100,
})

Moves up to limit messages off the dead-letter queue back to the queue each failed on, resets their retry, requeue and redelivery counters, sets their stored state back to pending, and returns how many were moved. Fix the cause first. See Retries and acks.

close()

Future<void> close()

Closes the broker, backend, event sink and control channel. Beat lock and state store are closed by Beat.stop.

See also: Architecture, Namespaces, Your first task.