Results and AsyncResult

A result backend stores each task’s state and return value under its id. Producers read it through AsyncResult. Without a backend, tasks are fire-and-forget.

import 'package:gisila_queue/gisila_queue.dart';

final r = await add.delay([2, 3]);
print(await r.state);                                   // TaskState.pending
print(await r.get(timeout: const Duration(seconds: 10))); // 5

AsyncResult

Handle to one enqueued task, mirroring Celery’s AsyncResult.

class AsyncResult<T> {
  final String id;
  Future<TaskState> get state;
  Future<bool> get ready;
  Future<bool> get successful;
  Future<T?> get({Duration timeout = const Duration(seconds: 30), Duration pollInterval = const Duration(milliseconds: 100)});
  Future<void> forget();
  Future<void> revoke({bool terminate = false});
}

When to use: to check on or wait for a task. Waiting inside a web request is fine for short tasks; for long ones, store id, return it to the client, and check later with app.resultFor(id).

Every member except revoke throws StateError if the app has no backend.

Member	Description
id	Task id and backend key.
state	Current TaskState; pending if nothing is stored (also after the result expired).
ready	true for a terminal state (success, failure, revoked).
successful	true for success.
get(timeout:, pollInterval:)	Polls until terminal, then returns the result cast to T?. Throws TaskFailureException on failure or revoked, and ResultTimeoutException after timeout.
forget()	Deletes the stored result.
revoke(terminate:)	Stores revoked (if there is a backend) and broadcasts a revoke on the control channel. Workers skip the task if they haven’t started it. A running task keeps running; terminate is carried but not enforced. See Control.

get casts the stored value with as T?. With RedisResultBackend the value has been through JSON, so T should be a JSON shape (int, String, Map<String, dynamic>, List<dynamic>): a task that returns a List<int> comes back as List<dynamic>, and AsyncResult<List<int>>.get() throws a cast error. InMemoryResultBackend returns the original object, so a test can pass where Redis fails.

try {
  final total = await r.get(timeout: const Duration(seconds: 5));
} on TaskFailureException catch (e) {
  print('task ${e.taskId} failed: ${e.message}');
} on ResultTimeoutException {
  print('still running; check later with app.resultFor(r.id)');
}

TaskState

enum TaskState { pending, scheduled, started, success, failure, retry, revoked }

Value	Set when
pending	Sent (no ETA), or nothing stored yet. Also set again by replayDeadLetters.
scheduled	Sent with a future ETA, or deferred by a rate limit.
started	A worker began running the handler.
success	The handler returned. Terminal.
failure	The handler failed with no retries left, timed out with none left, the task name was unknown after all requeues, or redeliveries were exceeded. Terminal.
retry	Re-enqueued for another attempt.
revoked	Revoked by a producer, expired, or rejected with RejectTask() (no requeue). Terminal.

Member	Description
isDone	true for success, failure, revoked.
TaskState.parse(String?)	By name; unknown or null gives pending.

ResultBackend

The abstract store.

abstract class ResultBackend {
  Future<void> store(String taskId, StoredResult result);
  Future<StoredResult?> fetch(String taskId);
  Future<void> forget(String taskId);
  Future<void> close() async {}

  Future<bool> chordArrive(String chordId, int total);
  Future<bool> chordAbort(String chordId);
  Future<String?> claimKey(String key, String taskId, Duration ttl);
}

When to use: implement it to store results elsewhere (a SQL table, for example). Override the three last methods with atomic operations if more than one process uses the backend; the default implementations keep state in the instance, which is only correct in one isolate.

Method	Description
store / fetch / forget	Write, read, delete one task’s StoredResult.
chordArrive(chordId, total)	Count one finished chord member. Returns true for exactly the arrival that reaches total (and never after an abort).
chordAbort(chordId)	Mark a chord aborted. Returns true only for the first caller.
claimKey(key, taskId, ttl)	Backs send(dedupeKey:). Returns null if the claim was made, or the id that already holds key.

StoredResult

One stored snapshot.

class StoredResult {
  StoredResult({required TaskState state, Object? result, String? error, DateTime? updatedAt});
  final TaskState state;
  final Object? result;
  final String? error;
  final DateTime? updatedAt;
}

toJson/fromJson and encode/decode (JSON by default) are what backends store. You read it from ResultBackend.fetch or a worker hook’s TaskObserver.

InMemoryResultBackend

Map-backed store for one isolate.

class InMemoryResultBackend extends ResultBackend {
  InMemoryResultBackend({int? maxEntries, Duration? ttl});
}

When to use: with InMemoryBroker in tests and single-process apps. It supports chords and dedupe within the process.

Parameter	Type	Default	Description
maxEntries	int?	null (unbounded)	Keep at most this many results; the least recently stored are evicted.
ttl	Duration?	null (never)	fetch treats results older than this (since last store) as gone.

RedisResultBackend

Redis store; each result is a string key with a TTL.

class RedisResultBackend extends ResultBackend implements Namespaced {
  RedisResultBackend({
    String host = 'localhost',
    int port = 6379,
    RedisOptions options = const RedisOptions(),
    String prefix = 'gq',
    Duration resultTtl = const Duration(hours: 24),
  });
}

When to use: any multi-process setup that reads results, uses dedupeKey, or runs chords. Use the same prefix as the broker.

Parameter	Type	Default	Description
host, port, options			Connection; see RedisOptions.
prefix	String	'gq'	Namespace.
resultTtl	Duration	24 h	Every store resets the key’s expiry to this. Also the safety TTL of chord counters.

Keys: <prefix>:result:<id>, <prefix>:dedupe:<task>:<key>, <prefix>:chord:<id>:count, <prefix>:chord:<id>:aborted. Chord arrival uses INCR, dedupe uses SET NX PX, so both are atomic across workers.

See also: Results concept, Getting results.