Events

Producers and workers can emit a lifecycle event for every task and worker state change. Events are off by default (NoopEventSink); attach a sink to feed GisilaRose, metrics, audit logs or alerts.

import 'package:gisila_queue/gisila_queue.dart';

final bus = InMemoryEventBus();
final app = GisilaQueue(broker: InMemoryBroker(), backend: InMemoryResultBackend(), events: bus);

bus.events().listen((e) {
  if (e.type == QueueEventType.taskFailed) print('${e.taskName} failed: ${e.fields['error']}');
});

QueueEventType

Value	Emitted by	Notable fields
taskSent	producer (send)	args, kwargs, retries, max_retries, eta, expires
taskReceived	worker, on reserve	args, kwargs
taskStarted	worker, before the handler
taskSucceeded	worker	runtime_ms, result
taskFailed	worker, terminal failure	error, runtime_ms (handler errors)
taskRetried	worker, retry or requeue	delay_ms, cause, or requeued: true, unknown_task: true
taskRevoked	worker: revoked, expired or RejectTask()	reason
taskRateLimited	worker, deferred by rate limit	delay_ms
workerOnline / workerHeartbeat / workerOffline	worker	queues, concurrency, active, succeeded, failed, retried, rate_limited, expired, revoked, dead_lettered, abandoned, paused_queues

Every worker task event also carries retries and max_retries.

Member	Description
isTask / isWorker	Category of the event.
QueueEventType.parse(String?)	By name; unknown gives taskSent.

QueueEvent

class QueueEvent {
  QueueEvent({
    required QueueEventType type,
    required DateTime timestamp,
    String? taskId,
    String? taskName,
    String? queue,
    String? workerId,
    String? hostname,
    Map<String, Object?>? fields,
  });
  num? numField(String key);
}

workerId and hostname are set on worker events and on task events emitted by a worker. toJson/fromJson and encode/decode use JSON (ts is ISO-8601 UTC).

QueueEventSink

Write side.

abstract class QueueEventSink {
  Future<void> emit(QueueEvent event);
  Future<void> close() async {}
}

When to use: implement it to forward events somewhere (Prometheus counters, a log pipeline). emit must never throw: observability must not break task processing.

Implementation	Description
NoopEventSink()	Drops events. Default.
MultiEventSink(List<QueueEventSink> sinks)	Fans out to several sinks.
InMemoryEventBus({int historySize = 1000})	Sink and source in one isolate; keeps the last historySize events for backfill.
RedisEventSink({host, port, options, prefix, historySize = 5000})	Publishes on <prefix>:events and keeps the latest historySize in the list <prefix>:events:history. Transport errors are swallowed.

QueueEventSource

Read side, used by monitors.

abstract class QueueEventSource {
  Stream<QueueEvent> events();
  Future<List<QueueEvent>> history({int limit = 500}) async => const [];
  Future<void> close() async {}
}

Implementation	Description
InMemoryEventBus	Same instance as the sink.
RedisEventSource({host, port, options, prefix})	Subscribes to <prefix>:events; history reads the capped list, oldest first.

final source = RedisEventSource(host: 'localhost', prefix: 'shop');
for (final e in await source.history(limit: 100)) {
  print('${e.timestamp} ${e.type.name} ${e.taskName}');
}
source.events().listen((e) => print(e));

See also: Monitoring, GisilaRose, Worker hooks.