Brokers

A broker moves TaskMessages from producers to workers. It holds delayed messages until due, hands ready ones to workers, and (on the distributed brokers) redelivers messages whose worker died before acking.

import 'package:gisila_queue/gisila_queue.dart';

Broker	Delivery	Processes	Delays	Use for
InMemoryBroker	at-most-once	one isolate	in-process list	tests, local dev, single-process background work
RedisBroker	at-least-once (visibility timeout)	many, many hosts	sorted set	most deployments; also powers beat lock, results, control, events
RabbitBroker	at-least-once (AMQP ack)	many, many hosts	TTL delay buckets + DLX	teams already running RabbitMQ

See Choosing a broker.

Broker

The abstract transport.

abstract class Broker {
  Future<void> enqueue(TaskMessage message);
  Future<TaskMessage?> reserve(List<String> queues, {required Duration timeout});
  Future<void> ack(TaskMessage message) async {}
  Duration? get visibilityTimeout => null;
  Future<void> touch(TaskMessage message) async {}
  Future<int> length(String queue);
  Future<int> purge(String queue);
  Future<void> close() async {}
}

When to use: implement it to add a transport, or subclass InMemoryBroker in tests to inject failures. Application code only calls length and purge directly.

Member	Description
enqueue(message)	Put the message on message.queue, or hold it until message.eta.
reserve(queues, timeout:)	Wait up to timeout for a ready message, checking queues in priority order. null on timeout.
ack(message)	Mark a reserved message done. No-op by default.
visibilityTimeout	How long a reservation may go unacked before redelivery, or null if never. Workers touch running messages every third of it.
touch(message)	Extend a running message’s reservation. No-op by default.
length(queue)	Ready (not delayed) messages on queue.
purge(queue)	Drop all ready messages on queue; returns the count. Delayed messages are untouched.
close()	Release connections.

InMemoryBroker

In-process broker backed by Dart collections.

class InMemoryBroker extends Broker {
  InMemoryBroker({Duration pollInterval = const Duration(milliseconds: 25)});
  int get scheduledCount;
}

When to use: tests, examples, and single-process apps that want background execution without Redis. Producers and workers must share the instance in one isolate. A message is removed when reserved and ack is a no-op, so a crash loses in-flight work (at-most-once).

Parameter / member	Type	Default	Description
pollInterval	Duration	25 ms	How often reserve re-checks while waiting. Lower it in tests.
scheduledCount	int		Messages waiting for a future ETA.

final app = GisilaQueue(
  broker: InMemoryBroker(pollInterval: const Duration(milliseconds: 5)),
  backend: InMemoryResultBackend(),
);

RedisBroker

Distributed broker on Redis.

class RedisBroker extends Broker implements Namespaced {
  RedisBroker({
    String host = 'localhost',
    int port = 6379,
    RedisOptions options = const RedisOptions(),
    String prefix = 'gq',
    Duration visibilityTimeout = const Duration(minutes: 5),
    int deadLetterLimit = 10000,
  });

  Duration get maintenanceInterval;
  Future<int> deadLetterCount();
  Future<int> unackedCount();
}

When to use: the default choice for production. Several worker processes on several hosts share the queues, and a crashed worker’s messages come back after visibilityTimeout. It targets a single Redis (or primary/replica), not Redis Cluster.

Parameter / member	Type	Default	Description
host / port	String / int	'localhost' / 6379	Redis address.
options	RedisOptions	RedisOptions()	Auth, TLS, database.
prefix	String	'gq'	Namespace for every key. Give each app its own.
visibilityTimeout	Duration	5 min	How long a reservation may go without ack or touch before it is redelivered. Workers touch running tasks every third of this, so it bounds crash-recovery time, not task runtime.
deadLetterLimit	int	10000	Cap on the <prefix>:dead list of undecodable messages.
maintenanceInterval	Duration		How often reserve promotes due delayed messages and redelivers stale ones: 1 s, or half the visibility timeout if shorter. Also the worst extra delay for an ETA.
deadLetterCount()	Future<int>		Messages on <prefix>:dead (undecodable payloads).
unackedCount()	Future<int>		Messages reserved but not yet acked, across all queues.

Keys it uses:

Key	Type	Purpose
<prefix>:queue:<name>	list	Ready messages, FIFO.
<prefix>:notify:<name>	list	Wake-up tokens; idle workers BLPOP these.
<prefix>:scheduled	sorted set	Delayed messages scored by due time.
<prefix>:unacked, <prefix>:unacked:payload	sorted set, hash	Reservation ledger keyed by delivery tag.
<prefix>:dead	list	Payloads that could not be decoded.

Each move between keys is one Lua script, so a crash leaves a message somewhere recoverable. The broker uses two connections (one for the blocking BLPOP) and reconnects and retries once when a call fails.

Note

<prefix>:dead only holds undecodable payloads. Tasks that failed go to the worker’s dead-letter queue, a normal queue (<prefix>:queue:dead_letter by default). See Dead letters.

final broker = RedisBroker(
  host: 'redis.internal',
  prefix: 'shop',
  visibilityTimeout: const Duration(minutes: 2),
);

RabbitBroker

Distributed broker on RabbitMQ (AMQP 0-9-1) via dart_amqp.

class RabbitBroker extends Broker implements Namespaced {
  RabbitBroker({
    String host = 'localhost',
    int port = 5672,
    String virtualHost = '/',
    String username = 'guest',
    String password = 'guest',
    String prefix = 'gq',
    int prefetchCount = 32,
  });

  static const minDelayBucketMs = 100;
  static const maxDelayBucketMs = 100 << 20;
  static int delayBucketFor(int delayMs);
  Future<void> deleteTopology();
}

When to use: you already operate RabbitMQ and want its durability and management UI. The distributed result backend, beat lock and state store, control channel and event transport only exist for Redis, so a multi-process RabbitMQ setup that needs any of them also runs Redis.

Parameter / member	Type	Default	Description
host, port, virtualHost		'localhost', 5672, '/'	Connection.
username / password	String	'guest' / 'guest'	PLAIN auth.
prefix	String	'gq'	Namespace for exchanges and queues.
prefetchCount	int	32	AMQP basic.qos prefetch: unacked deliveries RabbitMQ pushes to this process at once.
delayBucketFor(ms)	int		Largest bucket (100 ms × 2ⁿ) not longer than ms.
deleteTopology()	Future<void>		Deletes this prefix’s exchanges and queues. For test cleanup only.

Topology:

• Direct exchange <prefix>.tasks routes to durable queues <prefix>.queue.<name>.
• Delays use buckets <prefix>.delay.<ms> with fixed x-message-ttl and a dead-letter exchange back to <prefix>.tasks. A message enters the largest bucket that fits and hops to smaller ones until due, so long delays never block short ones and no plugin is needed. The longest bucket is about 29 hours; longer delays use it more than once.
• Messages are persistent and manually acked. Unacked deliveries (crash, or close() while in flight) are requeued by RabbitMQ.

There is no visibility timeout (visibilityTimeout is null): RabbitMQ redelivers when the connection drops. Undecodable payloads are rejected without requeue (dropped), unlike Redis which parks them.

final app = GisilaQueue(
  broker: RabbitBroker(host: 'rabbit', username: 'shop', password: 'secret', prefix: 'shop'),
  backend: RedisResultBackend(host: 'redis', prefix: 'shop'),
);

RedisOptions

Authentication and connection options shared by every Redis component (RedisBroker, RedisResultBackend, RedisControlChannel, RedisEventSink, RedisEventSource, RedisBeatLock, RedisBeatStateStore).

class RedisOptions {
  const RedisOptions({
    String? username,
    String? password,
    bool tls = false,
    int database = 0,
    SecurityContext? securityContext,
  });

  static ({String host, int port, RedisOptions options}) parseUrl(
    String url, {SecurityContext? securityContext});
}

When to use: whenever Redis needs a password, TLS or a non-zero database. Prefer parseUrl on a REDIS_URL environment variable.

Field	Type	Default	Description
username	String?	null	ACL user (Redis 6+). Sent only with a password.
password	String?	null	Sent with AUTH after connecting.
tls	bool	false	Connect over TLS.
database	int	0	Selected with SELECT when not 0.
securityContext	SecurityContext?	null	Trust roots and client certs for TLS.

parseUrl accepts redis://[user[:password]@]host[:port][/db] and rediss:// for TLS. A single credential is treated as the password, as redis-cli -u does. Missing host means localhost, missing port 6379. Other schemes throw FormatException.

import 'dart:io';

final r = RedisOptions.parseUrl(Platform.environment['REDIS_URL']!);
final broker = RedisBroker(host: r.host, port: r.port, options: r.options, prefix: 'shop');

Other exports

Symbol	Description
connectRedis(conn, host, port, options)	Opens a package:redis connection and applies TLS, AUTH and SELECT. Used by every Redis component.

See also: Redis vs RabbitMQ setup, Delivery guarantees.