Workflows (chain, group, chord)

Workflows compose Signatures (unevaluated calls) into pipelines and fan-outs. Build them with app.chain(), app.group() and app.chord(); nothing is sent until apply().

import 'package:gisila_queue/gisila_queue.dart';

final fetchUser = app.task<Map<String, dynamic>>('users.fetch', fetchUserHandler);
final greet = app.task<String>('users.greet', (ctx) async {
  final user = ctx.arg(0) as Map;
  return 'Hello, ${user['name']}';
});

final r = await app.chain<String>([fetchUser.sig(args: [42]), greet.sig()]).apply();
print(await r.get());

Signature

class Signature {
  const Signature(String taskName, {List<Object?> args = const [], Map<String, Object?> kwargs = const {}, String? queue, int maxRetries = 0});
  Signature withLeadingArg(Object? value);
  Map<String, Object?> toJson();
  factory Signature.fromJson(Map<String, Object?> json);
}

When to use: as the building block of every workflow. Prefer TaskRef.sig(), which fills in the task’s queue and maxRetries; the raw constructor defaults maxRetries to 0.

withLeadingArg(value) returns a copy with value as the first positional argument. That is how a chain passes a result to the next step.

Chain

Sequential pipeline: each step’s result becomes the next step’s first argument.

class Chain<T> {
  final List<Signature> signatures;
  Future<AsyncResult<T>> apply({String? queue, Duration? countdown, DateTime? eta});
}

When to use: steps that depend on the previous result (fetch, transform, store). For independent steps, use a group.

apply pre-assigns every step’s id, enqueues only the first step (carrying the rest in its message), and returns the AsyncResult of the last step. queue, countdown and eta apply to the first step only. When a step succeeds, the worker that ran it enqueues the next. When a step fails or is revoked, every remaining step is stored as failure (chain aborted: ...), so get() on the result throws instead of hanging.

A retrying step keeps the chain waiting; the next step starts after the retry succeeds.

Group

Parallel fan-out.

class Group<T> {
  final List<Signature> signatures;
  Future<GroupResult<T>> apply();
}

When to use: independent tasks whose results you want together (resize to several sizes). apply sends every signature at once.

GroupResult

class GroupResult<T> {
  final List<AsyncResult<T>> results;
  Future<List<T?>> get({Duration timeout = const Duration(seconds: 30), Duration pollInterval = const Duration(milliseconds: 100)});
  Future<bool> get ready;
}

get waits for all members and returns their results in order; the first failure is thrown. ready is true when every member is terminal.

final results = await app.group<int>([
  resize.sig(args: [imageId, 'small']),
  resize.sig(args: [imageId, 'large']),
]).apply();
final sizes = await results.get();

Chord

A group plus a callback that runs once, after every member succeeded.

class Chord<T> {
  final List<Signature> signatures;
  final Signature callback;
  Future<AsyncResult<T>> apply();
}

When to use: map-reduce shapes: run N tasks in parallel, then combine their results in one task, without a producer waiting in between.

The callback receives the list of member results (in signature order) as its first argument, followed by its own args. apply returns the callback’s AsyncResult. apply throws StateError if the app has no result backend: the barrier lives there (chordArrive/chordAbort). If any member fails or is revoked, the chord aborts once and the callback’s result is stored as failure (chord aborted: ...).

final sum = app.task<int>('sizes.sum', (ctx) async =>
    (ctx.arg(0) as List).fold<int>(0, (a, b) => a + (b as int)));

final total = await app.chord<int>(
  [resize.sig(args: [imageId, 'small']), resize.sig(args: [imageId, 'large'])],
  sum.sig(),
).apply();
print(await total.get());

Other exports

Symbol	Description
ChainLink	(id, signature) pair carried in TaskMessage.chain.
ChordBarrier	chordId, memberIds, callbackId, callback; carried in TaskMessage.chord.

See also: Build workflows, Workflows concept.