Workflows
What it is
Section titled “What it is”Workflows compose tasks without a producer waiting in the middle:
- chain: A then B then C, each getting the previous result as its first argument.
- group: A, B, C in parallel; wait for all.
- chord: a group, then one callback with all the results.
They are built from Signatures (taskRef.sig(...)).
When to use it
Section titled “When to use it”- A pipeline whose steps scale or retry independently (download, transcode, upload).
- Fan-out/fan-in: process N items in parallel and combine (sum sizes, merge reports).
- Long workflows where holding a web request or a worker slot while waiting would be wasteful.
When not to use it
Section titled “When not to use it”- Two quick steps that always run together: do both in one task.
- Dynamic, data-dependent graphs (the next step depends on a decision): have a task
enqueue the next task itself with
delay(). - When you need the intermediate results later: only the final step’s id is returned by a chain. Store intermediate data yourself, or use separate tasks.
How it works
Section titled “How it works”Chain. apply() pre-assigns an id to every step, enqueues the first step, and carries
the rest inside its message. The worker that finishes a step enqueues the next with the
result prepended to its arguments. If a step fails (after its retries) or is revoked,
every later step is marked failure with chain aborted: ..., so get() on the chain
throws instead of hanging. No backend is required to run a chain, but you need one to read
its result.
Group. apply() sends every signature at once and returns a
GroupResult. Nothing coordinates the members;
GroupResult.get() just waits for each.
Chord. Each member carries a barrier. When a member succeeds, its worker counts an
arrival in the backend; the arrival that completes the count fires the callback with all
member results. If any member fails or is revoked, the chord aborts once and the
callback’s result is stored as failure. Chords need a result backend
(RedisResultBackend across processes) and apply() throws StateError without one.
Arguments and results must be JSON-encodable, since they travel inside messages.
Example
Section titled “Example”import 'package:gisila_queue/gisila_queue.dart';
final download = app.task<String>('media.download', downloadHandler, maxRetries: 3);final transcode = app.task<String>('media.transcode', transcodeHandler);final publish = app.task<String>('media.publish', publishHandler);
final r = await app.chain<String>([ download.sig(args: ['https://example.com/video.mov']), transcode.sig(kwargs: {'format': 'mp4'}), // receives download's result as arg 0 publish.sig(),]).apply();
print(await r.get(timeout: const Duration(minutes: 10)));