README

A RabbitMQ client for PHP built on php-amqplib: declarative topology (exchanges, queues, bindings), fluent publish and consume, built-in retry with dead-lettering, message replay, a set of Symfony Console commands for operating queues, and a connector so phpdot/pool can hold and recycle connections.

Requirements
Installation
Usage
Architecture
Testing
License

Requirements

Requirement	Constraint
PHP	`>= 8.5`
`php-amqplib/php-amqplib`	`^3.0`
`phpdot/contracts`	`^0.1`
`psr/log`	`^3.0`
`symfony/console`	`^8.0`

php-amqplib brings ext-sockets and ext-mbstring. phpdot/container is a dev-only suggestion — the #[Config('rabbitmq')] attribute on RabbitMQConfig is inert until a phpdot application reflects it.

Installation

composer require phpdot/rabbitmq

Usage

Publish and consume

Topology is declared once in the config; the connection ensures it before publishing or consuming:

use PHPdot\RabbitMQ\RabbitMQConnection;
use PHPdot\RabbitMQ\Config\RabbitMQConfig;
use PHPdot\RabbitMQ\Message;
use PHPdot\RabbitMQ\Enum\TaskStatus;

$conn = new RabbitMQConnection(new RabbitMQConfig(
    host: 'localhost',
    exchanges: ['tasks' => ['type' => 'direct', 'durable' => true]],
    queues: [
        'tasks.process' => [
            'bindings' => [['exchange' => 'tasks', 'routing_key' => 'task.new']],
            'durable' => true,
        ],
    ],
));

$conn->message('{"task":"send_email"}')->publish('tasks', 'task.new');

$conn->consume('tasks.process')->execute(function (Message $msg): TaskStatus {
    processTask(json_decode($msg->body(), true));

    return TaskStatus::SUCCESS;
});

Returning TaskStatus::SUCCESS acks the message, RETRY requeues it through the retry flow, and FAILURE dead-letters it.

Retry and dead-lettering

A queue can declare a retry flow (delayed re-delivery) and a dead-letter target. Messages that exhaust their retries land on the dead-letter queue, where they can be inspected and replayed:

$result = $conn->replay('tasks.process.dead')
    ->limit(10)
    ->execute(); // or ->dryRun() to preview

CLI commands

The package ships Symfony Console commands for operating queues — rabbitmq:status, rabbitmq:queues, rabbitmq:topology:declare, rabbitmq:peek, rabbitmq:replay, and rabbitmq:dlq:analyze. Register them in your console application.

Pooling

RabbitMQConnector adapts a RabbitMQConnection to phpdot/pool's ConnectorInterface, so a pool can build, health-check, and recycle connections — one per coroutine.

Architecture

RabbitMQConnection owns a php-amqplib stream connection and channel, and drives topology declaration, resilient reconnection, and lifecycle. Publisher, Consumer, TopologyManager, and Replayer build on it; RabbitMQConnector bridges to phpdot/pool.

graph TD
    APP["Application / phpdot/pool"]
    CONNECTOR["RabbitMQConnector<br/><br/>Contracts Pool ConnectorInterface"]
    CONN["RabbitMQConnection<br/><br/>topology, resilient reconnect,<br/>channel lifecycle"]
    OPS["Publisher / Consumer<br/>TopologyManager / Replayer<br/><br/>publish, consume, retry + DLQ, replay"]
    CLI["Cli\\Command\\*<br/><br/>Symfony Console: status, queues,<br/>peek, replay, dlq:analyze"]
    AMQP["php-amqplib<br/><br/>AMQPStreamConnection + channel"]

    APP --> CONNECTOR
    CONNECTOR --> CONN
    APP --> OPS
    CLI --> OPS
    OPS --> CONN
    CONN --> AMQP

Testing

composer install
composer test        # PHPUnit
composer analyse     # PHPStan, level max + strict rules
composer cs-check    # PHP-CS-Fixer
composer check       # All three

The unit suite runs with no broker. The integration suite connects to a RabbitMQ at localhost:5672 and skips automatically when none is reachable.

License

MIT.

This repository is a read-only mirror, generated by CI from phpdot/monorepo. Pull requests and issues belong in the monorepo.

phpdot / rabbitmq

Maintainers

Package info

Statistics

Security