Download the PHP package quellabs/canvas-scheduler-rabbitmq without Composer

canvas-rabbitmq

RabbitMQ queue consumer and job dispatcher for the Canvas PHP framework.

Adds a RabbitMQ-backed job queue to Canvas's Task Scheduler system. Jobs are dispatched from application code via dependency injection and processed by a long-running worker process.

Requirements

• PHP 8.2+
• Canvas 1.x
• RabbitMQ server
• php-amqplib (installed automatically)

Installation

No further configuration is required. The package registers itself automatically via Canvas's service discovery.

Configuration

Set the default queue driver in config/app.php:

This tells Canvas to inject RabbitMQQueue by default when QueueInterface is requested. If you only have one queue package installed, this can be omitted.

To request a specific driver regardless of the default, use contextual DI:

RabbitMQ-specific settings go in config/scheduler-rabbitmq.php:

exchange_name defaults to the empty string, which uses RabbitMQ's built-in default exchange and routes messages directly to the queue by name. Set this only if you are using a custom exchange topology.

prefetch_count controls how many unacknowledged messages each worker holds at once. The default of 1 gives the fairest work distribution when running multiple worker processes.

Creating a Job

Implement QueueableInterface on any class. Constructor parameters become the serializable payload:

Payload keys must match constructor parameter names exactly — the worker reconstructs the job via Canvas's DI container using make($class, $payload).

getTimeout() returns the maximum number of seconds the job is expected to run. This value is stored in the job envelope for observability but is not enforced by the worker itself. To enforce it, set stopwaitsecs in your Supervisord configuration to the longest expected job duration across all job types (see below).

Dispatching Jobs

Inject QueueInterface into any controller or service:

The job is published to RabbitMQ immediately and processed by the worker when it is next available.

Running the Worker

Start the worker via Sculpt:

The worker polls RabbitMQ for jobs until it reaches the configured queue_max_jobs limit (default: 500), then exits cleanly. Use Supervisord to keep it running:

stopwaitsecs should be set to the longest expected job duration across all job types. Supervisord will send SIGKILL to any worker that has not stopped within this window after receiving SIGTERM.

Restart workers after deployment so they pick up new code:

Retry and Failure Handling

Failed jobs are retried up to getMaxRetries() times. On each failure the original message is rejected and a fresh message with an incremented attempt counter is published back to the pending queue. Jobs that exhaust all retries are published to a separate failed queue for inspection.

Queue Layout

Both queues are declared durable and messages are published as persistent, so jobs survive a RabbitMQ broker restart.

License

MIT