Build workflows

You’ll build an image pipeline: fetch an image, resize it to several sizes in parallel, then record the total size, using a chain, a group and a chord.

Prerequisites

An app with a result backend (chords require one; reading any workflow result does too).
A worker that registers all the tasks.

1. Register the tasks

import 'package:gisila_queue/gisila_queue.dart';

void registerImageTasks(GisilaQueue app) {
  app.task<String>('images.fetch', (ctx) async => downloadToStorage(ctx.arg(0) as String));
  app.task<int>('images.resize', (ctx) async {
    final key = ctx.arg(0) as String;
    final size = ctx.arg(1) as String;
    return resizeAndStore(key, size); // returns bytes written
  }, maxRetries: 2, autoRetry: true);
  app.task<int>('images.total', (ctx) async {
    final sizes = (ctx.arg(0) as List).cast<int>();
    final total = sizes.fold<int>(0, (a, b) => a + b);
    await recordUsage(ctx.arg(1) as String, total);
    return total;
  });
}

Get TaskRefs by keeping the return values of app.task, or build signatures by name with Signature('images.resize', args: [...], maxRetries: 2).

2. Chain: fetch, then process

final fetch = app.task<String>('images.fetch', fetchHandler);
final makeSmall = app.task<int>('images.resize_small', resizeSmallHandler);

final r = await app.chain<int>([
  fetch.sig(args: ['https://example.com/cat.jpg']),
  makeSmall.sig(), // gets fetch's result (the storage key) as arg 0
]).apply(queue: 'images');

print(await r.get(timeout: const Duration(minutes: 2)));

apply(queue:, countdown:, eta:) affects the first step only. Later steps use their signature’s queue.

3. Group: run in parallel

final group = await app.group<int>([
  resize.sig(args: [key, 'small']),
  resize.sig(args: [key, 'medium']),
  resize.sig(args: [key, 'large']),
]).apply();

final sizes = await group.get(timeout: const Duration(minutes: 2)); // [s, m, l]

4. Chord: parallel, then combine

final total = await app.chord<int>(
  [
    resize.sig(args: [key, 'small']),
    resize.sig(args: [key, 'medium']),
    resize.sig(args: [key, 'large']),
  ],
  totalTask.sig(args: [accountId]), // called as images.total([s, m, l], accountId)
).apply();

print('stored ${await total.get()} bytes');

The producer can exit right after apply(); the worker that finishes the last member enqueues the callback.

5. Handle failure

If any chain step or chord member fails after its retries (or is revoked), the remaining chain steps or the chord callback are marked failure immediately:

try {
  await total.get();
} on TaskFailureException catch (e) {
  print(e.message); // chord aborted: member task "images.resize" failure (...)
}

Verify it works

Run a worker with all three tasks registered and enqueue the chord. The worker log shows three images.resize ✓ lines, then images.total ✓, and total.get() returns the sum.