Control channel

The control channel broadcasts instructions from any producer to every running worker: revoke a task, pause a queue, resume a queue. It is Celery’s app.control.

import 'package:gisila_queue/gisila_queue.dart';

final app = GisilaQueue(
  broker: RedisBroker(prefix: 'shop'),
  backend: RedisResultBackend(prefix: 'shop'),
  controlChannel: RedisControlChannel(prefix: 'shop'),
);

await app.controlChannel.publish(const ControlMessage.pause('reports'));
await app.controlChannel.publish(const ControlMessage.resume('reports'));

`ControlChannel`

abstract class ControlChannel {
  Future<void> publish(ControlMessage message);
  Stream<ControlMessage> subscribe();
  Future<void> close() async {}
}

When to use: to enable AsyncResult.revoke across processes and to pause queues during an incident or a downstream outage. Leave the default NoopControlChannel when you need neither.

There is no history: a worker only sees messages published after it subscribed (at run()). Revokes are also written to the result backend, so a worker that starts later still skips a revoked task if a backend is configured. Pause and resume are not persisted: a restarted worker consumes all its queues again.

Implementation	Description
`NoopControlChannel()`	Drops everything. Default on `GisilaQueue`.
`InMemoryControlChannel()`	Broadcast stream in one isolate; producers and workers share the instance.
`RedisControlChannel({host, port, options, prefix})`	Redis pub/sub on `<prefix>:control`. Reaches every subscribed worker process. Added by `GisilaQueue.redis` (with `control: true`) and by `WorkerCli`.

`ControlMessage`

class ControlMessage {
  const ControlMessage.revoke(String taskId, {bool terminate = false});
  const ControlMessage.pause(String queue);
  const ControlMessage.resume(String queue);

  final ControlMessageType type;
  final String? taskId;
  final String? queue;
  final bool terminate;
}

Constructor	Worker effect
`revoke(taskId)`	Remembers the id for an hour; skips the task if reserved later (records `revoked`). A handler already running sees `ctx.isRevoked == true` but is not stopped.
`pause(queue)`	Stops reserving from `queue`. Running tasks continue.
`resume(queue)`	Starts reserving from `queue` again.

terminate is carried on the wire but not enforced: a worker cannot stop a running handler inside its isolate.

toJson/fromJson and encode/decode serialize to JSON; fromJson throws FormatException for unknown types.

`ControlMessageType`

enum ControlMessageType { revoke, pause, resume }

See also: Monitoring (Rose exposes revoke and pause/resume over HTTP), Workers.