Producer API

Producers enqueue work through a TaskRef (returned by app.task()), or by name through GisilaQueue.send. Every call returns an AsyncResult right after the message is on the broker; it does not wait for a worker.

import 'package:gisila_queue/gisila_queue.dart';

final r = await add.delay([2, 3]);
await add.applyAsync(args: [10, 20], queue: 'math', countdown: const Duration(seconds: 30));

TaskRef

Typed producer handle for a registered task. T is the handler’s result type, surfaced by AsyncResult<T>.get().

class TaskRef<T> {
  final Task task;
  String get name;

  Future<AsyncResult<T>> delay([List<Object?> args = const [], Map<String, Object?> kwargs = const {}]);

  Future<AsyncResult<T>> applyAsync({
    List<Object?> args = const [],
    Map<String, Object?> kwargs = const {},
    String? queue,
    Duration? countdown,
    DateTime? eta,
    Duration? expiresIn,
    DateTime? expires,
  });

  Signature sig({List<Object?> args = const [], Map<String, Object?> kwargs = const {}, String? queue, int? maxRetries});
}

When to use: whenever the producer has the TaskRef in scope. It carries the task’s maxRetries into each message, which app.send() does not.

delay()

Shorthand for applyAsync(args: args, kwargs: kwargs). Use it for the common “run this soon on its usual queue” case.

applyAsync()

Parameter	Type	Default	Description
args	List<Object?>	[]	Positional arguments. Must be JSON-encodable.
kwargs	Map<String, Object?>	{}	Keyword arguments. Must be JSON-encodable.
queue	String?	null	Explicit queue. Always wins over router, task default and app default.
countdown	Duration?	null	Run no earlier than now + countdown. Wins over eta if both are set.
eta	DateTime?	null	Run no earlier than this time.
expiresIn	Duration?	null	Latest useful run time, relative. Wins over expires.
expires	DateTime?	null	Latest useful run time, absolute. A worker that picks the message up later records it as revoked instead of running it.

A delayed message is stored as scheduled in the backend; otherwise pending. Unencodable arguments throw UnencodableArgumentsException before anything is sent.

await sendReminder.applyAsync(
  args: [userId],
  eta: DateTime.now().add(const Duration(days: 1)),
  expiresIn: const Duration(days: 2),
);

sig()

Builds a Signature (an unevaluated call) for chains, groups and chords. queue defaults to the task’s queue and maxRetries to the task’s maxRetries.

GisilaQueue.send()

Enqueues by task name.

Future<AsyncResult<T>> 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),
})

When to use: when the producer doesn’t register the task itself (a web process that only knows names), or when you need dedupeKey. Pass maxRetries yourself: it does not read the task’s setting. id, chain and chord are used internally by workflows; leave them unset.

Parameter	Type	Default	Description
taskName	String	required	Registered name. The producer doesn’t need it registered; the worker does.
args, kwargs, queue, eta, expires			As on applyAsync.
maxRetries	int	0	Retry ceiling carried in the message.
dedupeKey	String?	null	Idempotency key. If a send with the same task name and key happened within dedupeFor, nothing is enqueued and the earlier task’s AsyncResult is returned. Needs a result backend, otherwise StateError.
dedupeFor	Duration	1 hour	How long the key is held.

Deduplication is atomic across processes with RedisResultBackend (SET NX PX). The in-memory backend keeps claims per instance.

// Clicking "pay" twice enqueues one charge.
final r = await app.send<void>(
  'billing.charge',
  args: [orderId],
  maxRetries: 3,
  dedupeKey: 'order-$orderId',
);

Queue resolution

For every send, the queue is the first of:

1. the explicit queue: argument,
2. the app’s Router match,
3. the task’s registered queue (only if the task is registered in this process),
4. GisilaQueue.defaultQueue.

Router

Declarative task-name to queue routing, Celery’s task_routes.

class Router {
  void route(String glob, {required String queue});
  void routeMatching(RegExp pattern, {required String queue});
  String? resolve(String taskName);
}

When to use: to keep queue names out of call sites, for example routing every emails.* task to an emails queue that its own workers consume. For one-off overrides, pass queue: at the call.

Method	Description
route(glob, queue:)	* matches any run of characters; everything else is literal. The glob is anchored, so 'emails.*' matches emails.send but not x.emails.send.
routeMatching(pattern, queue:)	Uses the RegExp as given (add ^/$ yourself).
resolve(taskName)	Queue of the first matching rule in insertion order, or null.

final router = Router()
  ..route('emails.*', queue: 'emails')
  ..routeMatching(RegExp(r'^reports\.'), queue: 'reports');

final app = GisilaQueue(broker: RedisBroker(prefix: 'shop'), router: router);

Note

package:gisila/gisila.dart exports shelf_router’s Router. In a file that imports both, hide one: import 'package:gisila_queue/gisila_queue.dart' hide Router, LogLevel; or use a prefix.

See also: Route tasks to queues, Schedule delayed tasks, Messages and codec.