GisilaRose dashboard

GisilaRose is an optional dashboard that folds the event stream into live task, worker and queue views, and can fire, retry, revoke and pause. It lives in a separate library so the core doesn’t pull in an HTTP server.

import 'package:gisila_queue/gisila_queue.dart';
import 'package:gisila_queue/gisila_rose.dart';

GisilaRose

class GisilaRose {
  GisilaRose({
    required QueueEventSource source,
    Broker? broker,
    ControlChannel? control,
    QueueEventSink? events,
    ResultBackend? backend,
    String? basicAuth,
    MonitorState? state,
    String host = '127.0.0.1',
    int port = 5555,
    bool allowUnauthenticated = false,
    Duration queuePollInterval = const Duration(seconds: 2),
    Uuid? idGenerator,
  });

  int? get boundPort;
  Future<void> start();
  Future<void> pollQueuesNow();
  Future<void> stop();
}

When to use: to see what the cluster is doing and act on it during development or incidents. For alerting and long-term metrics, write your own QueueEventSink instead; Rose keeps state in memory only. The bundled rose binary covers most setups (see CLI).

Parameter / member	Type	Default	Description
source	QueueEventSource	required	Event stream to fold (RedisEventSource or a shared InMemoryEventBus).
broker	Broker?	null	Enables queue depths, fire and retry.
control	ControlChannel?	null	Enables revoke, pause, resume.
events	QueueEventSink?	null	Receives taskSent when Rose fires or retries.
backend	ResultBackend?	null	Stores pending for tasks Rose enqueues.
basicAuth	String?	null	'user:pass'; every route then requires HTTP Basic Auth.
state	MonitorState?	MonitorState()	The in-memory model.
host / port	String / int	'127.0.0.1' / 5555	Bind address. port: 0 picks a free port (see boundPort).
allowUnauthenticated	bool	false	Allow write actions without auth on a non-loopback host.
queuePollInterval	Duration	2 s	How often queue depths are refreshed.
start()			Backfills history, binds the server, starts polling. Throws StateError if a broker or control channel is set, host is not loopback, and there is neither basicAuth nor allowUnauthenticated.
pollQueuesNow()			Refresh depths immediately.
stop()			Stops serving and closes source, broker, control, events and backend.

Rose polls the depth of every queue a worker reported plus every queue it fired to. Fire uses queue default unless the request names one (the app’s router is not consulted), and maxRetries 0 unless max_retries is in the body. Retry re-sends the task’s name, args, kwargs, queue and max_retries under a new id. Rose sends no CORS headers, and every POST (including revoke and pause, which have no body) must carry Content-Type: application/json or it gets 415. Auth failures get 401.

final rose = GisilaRose(
  source: RedisEventSource(host: 'localhost', prefix: 'shop'),
  broker: RedisBroker(host: 'localhost', prefix: 'shop'),
  control: RedisControlChannel(host: 'localhost', prefix: 'shop'),
  events: RedisEventSink(host: 'localhost', prefix: 'shop'),
  backend: RedisResultBackend(host: 'localhost', prefix: 'shop'),
  basicAuth: 'admin:secret',
  host: '0.0.0.0',
);
await rose.start();

HTTP API

Endpoint	Returns / action	Needs
GET /	HTML dashboard
GET /api/stats	Totals, worker counts, queue depths, per-task-name stats, capabilities
GET /api/workers	Each worker with counters and active_task_ids
GET /api/workers/<id>	One worker with its recent tasks
GET /api/tasks	Recent tasks; query state, name, queue, worker, limit, offset
GET /api/tasks/<id>	One task’s detail
POST /api/tasks/fire	Enqueue {name, args?, kwargs?, queue?, max_retries?}; returns {ok, id}	broker
POST /api/tasks/<id>/retry	Re-fire with a new id	broker
POST /api/tasks/<id>/revoke	Broadcast revoke	control
POST /api/queues/<name>/pause	Pause a queue	control
POST /api/queues/<name>/resume	Resume a queue	control
GET /api/queues	Depth per known queue
GET /api/events	Server-Sent Events stream of live events

Mutating endpoints return 503 when their dependency isn’t configured.

MonitorState

The in-memory model Rose renders.

class MonitorState {
  MonitorState({int maxTasks = 10000, Duration workerTimeout = const Duration(seconds: 30)});
}

When to use: embed it to build your own monitor or tests: bind(source) (backfill then live) or apply(event) by hand, then query.

Member	Description
maxTasks	Tasks retained; oldest evicted.
workerTimeout	A worker with no heartbeat for this long counts as not alive.
sent, received, succeeded, failed, retried, revoked, rateLimited	Lifetime counters.
queueLengths	Latest known depth per queue.
tap	Stream of every live applied event.
bind(source) / apply(event, {broadcast})	Feed events.
tasks({state, name, queue, worker, limit = 200, offset = 0})	Filtered, newest first.
tasksForWorker(id, {state}), activeTaskIdsFor(id)	Per-worker views.
taskById(id), workerById(id), taskCount, workers	Lookups.
taskTypeStats(), stats(now, {capabilities})	Aggregates (the /api/stats payload).
close()	Cancel the subscription.

TaskInfo and WorkerInfo

Mutable views folded from events. TaskInfo has id, name, queue, workerId, state, args, kwargs, result, error, retries, maxRetries, runtimeMs, sentAt, receivedAt, startedAt, finishedAt, updatedAt. WorkerInfo has id, hostname, queues, concurrency, active, succeeded, failed, retried, online, lastHeartbeat and isAlive(now, timeout). Both have toJson().

See also: Monitor your queues.