README

kinetis/queue-rabbitmq
A Fiber-native, non-blocking RabbitMQ backend for kinetis/queue's QueueInterface

Part of Kinetis, a non-blocking PHP framework for API-first applications, developed in the kinetis-dev/kinetis monorepo.

Adds RabbitMQ as a queue backend. push()/pop()/ack()/fail() work exactly like any other backend — only your configuration changes. release() does too, with one difference worth knowing: it's two separate AMQP operations rather than one atomic step, so a crash between them can redeliver a job twice. See kinetis.dev/docs/queue-rabbitmq.html for why, and what other backends don't share this.

Delays are broker-driven and independent of each other: a job delayed by three seconds waits three seconds, not the hour an earlier delayed job on the same queue still has to go. A delay is a floor — the job is available no sooner than that, and the broker delivers it when it gets to it. Nothing beyond a stock RabbitMQ is needed — no plugin. A delayed retry travels the same ladder: release() publishes its replacement into it rather than onto the real queue, still confirmed before the original delivery is discarded. Delays cap at 4,194,303 seconds (about 48 days), the longest queue TTL the AMQP client can encode, and a longer one is rejected by push() and release() alike before anything is sent.

use Kinetis\Config\Config;
use Kinetis\QueueRabbitMq\RabbitMqQueueFactory;

$queue = RabbitMqQueueFactory::fromConfig($config);

$queue->push(new SendWelcomeEmail($email, $name), queue: 'default');

The factory builds the queue's own Thesis\Amqp\Client and hands the queue that client's disconnect(), so RabbitMqQueue declares Kinetis\Queue\DisposableQueueInterface and dispose() closes the channel and the connection when the worker ends. Building one yourself means registering that — $app->onDispose($queue->dispose(...)); the bootstrap behind QUEUE_CONNECTION=rabbitmq already does. A RabbitMqQueue constructed directly around a client you built disconnects nothing: that client stays yours.

RabbitMqQueue declares Kinetis\Queue\ClearableQueueInterface, purging the queue and every delay tier and reporting the total the broker says it removed. queue.purge leaves messages already delivered to a consumer and not yet acked in place — the broker's own rule, which happens to be exactly the contract's.

A delivery tag is scoped to its channel, and reusing one is a channel-level protocol error rather than an answer this package can read back, so it raises no Kinetis\Queue\Exception\StaleJobHandleException. An unacked delivery is requeued as soon as the connection drops, so a worker that dies mid-job has its work redelivered and handlers have to be idempotent.

Configuration

QUEUE_CONNECTION=rabbitmq
QUEUE_RABBITMQ_URL=amqp://guest:guest@localhost:5672/

Both are scoped — QUEUE_RABBITMQ_URL + events → QUEUE_EVENTS_RABBITMQ_URL. kinetis/queue's own keys (QUEUE_CONNECTION, QUEUE_MAX_ATTEMPTS, ...) are documented in that package; full reference: kinetis.dev/docs/config.html.

The URI's username, password and vhost are percent-decoded, so amqp://guest:p%40ssword@rabbit:5672/%2Fstaging authenticates as guest with the password p@ssword against the vhost /staging.

A queue name resolves directly to a RabbitMQ queue of that name, declared durable the first time anything touches it — nothing to create ahead of time. Delayed jobs additionally use queues and exchanges named {queue}.delay.{seconds}s, declared the same way; a queue name can't contain a ., so none of them can collide with a queue of your own.

Installation

composer require kinetis/queue-rabbitmq

Requires PHP 8.4+, kinetis/framework, and kinetis/queue. Full documentation: kinetis.dev/docs/queue-rabbitmq.html.

License

MIT — see LICENSE.