kinetis / queue-rabbitmq

v1.3.0 2026-09-05 08:02 UTC

Requires

• php: ^8.4
• amphp/amp: ^3.1.3
• kinetis/framework: ^1.3.1
• kinetis/queue: ^1.2.0
• thesis/amqp: ^1.0.2

Requires (Dev)

• infection/infection: ^0.35.0
• phpstan/phpstan: ^2.2.8
• phpunit/phpunit: ^12.5.33
• vimeo/psalm: ^6.16.1

MIT

• Alon Noy

Published on Packagist.org on 2026-09-05 08:08 UTC

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. 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 at push().

use Kinetis\Config\Config;
use Kinetis\QueueRabbitMq\RabbitMqClientFactory;
use Kinetis\QueueRabbitMq\RabbitMqQueue;

$queue = new RabbitMqQueue(RabbitMqClientFactory::fromConfig($config));

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

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/

Key	Default	Purpose
QUEUE_RABBITMQ_URL	(required)	amqp:// URI.
QUEUE_RABBITMQ_QUEUE_PREFIX	—	Prepended to every queue name.

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.

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.