kinetis/queue-rabbitmq

A Fiber-native non-blocking RabbitMQ (AMQP 0-9-1) backend for kinetis/queue's QueueInterface.

v1.0.3 2026-08-16 08:57 UTC

Requires

• php: ^8.4
• amphp/amp: ^3.1
• kinetis/framework: ^1.6
• kinetis/queue: ^1.2
• thesis/amqp: ^1.0
• thesis/time-span: ^0.2

Requires (Dev)

• infection/infection: ^0.34.2
• kinetis/cache-redis: ^1.0
• kinetis/persistence: ^1.7
• phpstan/phpstan: ^2.2
• phpunit/phpunit: ^11.0
• vimeo/psalm: ^6.16

MIT

• Alon Noy

Published on Packagist.org on 2026-08-16 08:57 UTC

README

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

Adds RabbitMQ as a queue backend. push()/pop()/ack()/release()/fail() work exactly like any other backend — only your configuration changes.

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');

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: docs.kinetis.dev/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. Don't name a queue ending in .delay; that suffix is reserved for the internal queue delayed jobs route through.

Important: opening this connection disables concurrently() in that process

Once anything calls push()/pop() for the first time, Kinetis\Async\concurrently() can't be called again anywhere in that same OS process, for any reason, for as long as the connection stays open — which is indefinitely, since nothing here closes it on its own. RabbitMQ keeps a connection open and listening at all times, and concurrently() waits for everything pending in the process to settle before it returns, which never happens while that connection stays open.

This never affects the kinetis queue:work loop itself. It does affect two other things:

• A job's own handle() reaching for concurrently() for its own unrelated work, once any RabbitMqQueue in that process has opened a connection.
• A persistent HTTP worker (FrankenPHP), not just a queue worker. If a controller calls push() to enqueue a job, that opens the connection in the request-handling worker process too — and a persistent worker keeps running that same process across many unrelated requests afterward. Every later request that process happens to serve loses the ability to call concurrently(), even if that request never touches this queue at all, until the worker restarts.

Installation

composer require kinetis/queue-rabbitmq

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

License

MIT — see LICENSE.