README

RabbitMQ integration for PHP applications, built on php-amqplib/php-amqplib. Includes typed JSON messages using Valinor and Nette Utils, base producers and consumers, queue declaration, consumer workers and Symfony Console commands.

Installation

Requires PHP 8.4.1 or newer, with the mbstring and sockets extensions. The CI matrix covers PHP 8.4 and 8.5.

composer require mistrfilda/amqp

The package uses the Mistrfilda\Amqp namespace. It includes Symfony Console 8, Valinor 2 and Nette Utils 4. It does not require a full Nette or Symfony application.

Messages, producers and consumers

Implement RabbitMQMessage for each message DTO. The application supplies the request ID and enqueue timestamp; the library serializes them with the other message fields. Keep the DTO compatible with messages already waiting in a queue.

For example, place these classes in separate files under your application's App\Messaging namespace:

namespace App\Messaging;

use Mistrfilda\Amqp\BaseConsumer;
use Mistrfilda\Amqp\BaseProducer;
use Mistrfilda\Amqp\RabbitMQMessage;

final readonly class ExampleMessage implements RabbitMQMessage
{
	public function __construct(
		public string $requestId,
		public int $messageQueuedAtTimestamp,
		public string $value,
	) {
	}

	public function getRequestId(): string
	{
		return $this->requestId;
	}

	public function getMessageQueuedAtTimestamp(): int
	{
		return $this->messageQueuedAtTimestamp;
	}
}

/** @extends BaseProducer<ExampleMessage> */
final class ExampleProducer extends BaseProducer
{
	protected function getMessageClass(): string
	{
		return ExampleMessage::class;
	}
}

/** @extends BaseConsumer<ExampleMessage> */
final class ExampleConsumer extends BaseConsumer
{
	protected function processMessage(RabbitMQMessage $messageObject): void
	{
		echo $messageObject->value, PHP_EOL;
	}

	protected function getMessageClass(): string
	{
		return ExampleMessage::class;
	}
}

BaseProducer::publish() validates the message class before serializing it. BaseConsumer::consume() maps the JSON payload into the configured DTO using Valinor's permissive type mapping, calls processMessage() and returns Ack only when processing succeeds. Mapping and processing exceptions propagate to the worker.

Configure and use a queue

Supply $host, $port, $user and $password from your application's configuration. Constructing the connection factory does not connect to RabbitMQ; it connects when a channel is first requested and reuses the connection for subsequent channels.

use App\Messaging\ExampleConsumer;
use App\Messaging\ExampleMessage;
use App\Messaging\ExampleProducer;
use Mistrfilda\Amqp\PhpAmqpLibRabbitMQPublisher;
use Mistrfilda\Amqp\RabbitMQConnectionConfig;
use Mistrfilda\Amqp\RabbitMQConnectionFactory;
use Mistrfilda\Amqp\RabbitMQConsumerRunner;
use Mistrfilda\Amqp\RabbitMQQueueConfig;
use Mistrfilda\Amqp\RabbitMQQueueConfigCollection;
use Mistrfilda\Amqp\RabbitMQQueueDeclarator;

$connectionFactory = new RabbitMQConnectionFactory(
	new RabbitMQConnectionConfig($host, $port, $user, $password),
);
$queues = new RabbitMQQueueConfigCollection([
	new RabbitMQQueueConfig(
		name: 'exampleQueue',
		consumerName: 'exampleConsumer',
		consumer: new ExampleConsumer(),
		prefetchCount: 5,
		durable: true,
		requeueOnFailure: true,
	),
]);
$declarator = new RabbitMQQueueDeclarator($connectionFactory, $queues);
$runner = new RabbitMQConsumerRunner($connectionFactory, $queues);
$publisher = new PhpAmqpLibRabbitMQPublisher($connectionFactory);
$producer = new ExampleProducer($publisher, 'exampleQueue');

try {
	$declarator->declareQueues();
	$producer->publish(new ExampleMessage('request-123', time(), 'Hello'));
	$runner->consume('exampleConsumer', secondsToLive: 60);
} finally {
	$connectionFactory->close();
}

Queue declaration, publishing and consumption can run in separate processes. Declare queues before publishing or starting a worker. Publishing does not create queues automatically. Queues are non-exclusive and are not auto-deleted; durability is configurable. Each consumer name should uniquely identify its queue configuration.

The initial connection configuration uses the default virtual host / and the php-amqplib connection defaults. It accepts host, port, user and password.

Console commands

use Mistrfilda\Amqp\RabbitMQConsumerCommand;
use Mistrfilda\Amqp\RabbitMQDeclareQueuesCommand;

$console->addCommands([
	new RabbitMQDeclareQueuesCommand($declarator),
	new RabbitMQConsumerCommand($runner),
]);

For an application whose console entry point is bin/console:

php bin/console rabbitmq:declare-queues
php bin/console rabbitmq:consumer exampleConsumer
php bin/console rabbitmq:consumer exampleConsumer 3600

The optional secondsToLive argument must resolve to a positive number of seconds. Omitting it runs until the consumer stops or an exception escapes. The time limit is checked between deliveries; it does not interrupt an executing handler.

Nette DI example

The library can be registered through ordinary service definitions. Define the rabbitmq.host, rabbitmq.port, rabbitmq.user and rabbitmq.password parameters in your application's configuration. Command discovery belongs to your console integration, for example Contributte Console.

services:
	rabbitMQConnectionConfig: Mistrfilda\Amqp\RabbitMQConnectionConfig(%rabbitmq.host%, %rabbitmq.port%, %rabbitmq.user%, %rabbitmq.password%)
	rabbitMQConnectionFactory: Mistrfilda\Amqp\RabbitMQConnectionFactory
	rabbitMQPublisher: Mistrfilda\Amqp\PhpAmqpLibRabbitMQPublisher
	exampleConsumer: App\Messaging\ExampleConsumer
	exampleProducer: App\Messaging\ExampleProducer(@rabbitMQPublisher, 'exampleQueue')
	rabbitMQQueueConfigCollection: Mistrfilda\Amqp\RabbitMQQueueConfigCollection([
		Mistrfilda\Amqp\RabbitMQQueueConfig('exampleQueue', 'exampleConsumer', @exampleConsumer, prefetchCount: 5),
	])
	rabbitMQQueueDeclarator: Mistrfilda\Amqp\RabbitMQQueueDeclarator
	rabbitMQConsumerRunner: Mistrfilda\Amqp\RabbitMQConsumerRunner
	- Mistrfilda\Amqp\RabbitMQDeclareQueuesCommand
	- Mistrfilda\Amqp\RabbitMQConsumerCommand

Delivery and failure behaviour

Implement RabbitMQConsumerHandler directly when you need to choose a result explicitly or consume a payload without the base JSON mapper.

Handler outcome	Worker behaviour
`RabbitMQConsumeResult::Ack`	Acknowledge the delivery.
`RabbitMQConsumeResult::Nack`	Negatively acknowledge and requeue the delivery.
`RabbitMQConsumeResult::Reject`	Reject without requeueing.
Exception	Negatively acknowledge using `requeueOnFailure`, then rethrow and stop the worker.

requeueOnFailure only affects exceptions. An explicit Nack always requeues. Messages rejected without requeueing may be dead-lettered if the broker is configured for it. Queue configuration defaults to prefetch 5, durable queues and requeueing on failure.

The publisher sends persistent messages with application/json content type via the default exchange, using the queue name as the routing key. Its optional headers argument sets AMQP application headers. Channels are closed after publishing, declaring queues or finishing a worker, including on failure.

There are no publisher confirms, automatic publish retries, delayed retries or dead-letter topology management. A successful publish() call is not a broker confirmation. Applications own idempotency, logging, database transactions and worker supervision.

Development

composer install
composer cs-fix
composer build-all

build-all runs strict Composer validation, PHP syntax checks, PHPStan at max for src and tests with strict/deprecation/PHPUnit rules and Valinor support, the Slevomat-based coding standard, and PHPUnit. Individual scripts are lint, stan, cs, cs-fix and test-unit.

Tests use test doubles and do not connect to RabbitMQ or a database. Development dependencies and caches are excluded from version control. The library does not commit composer.lock, so CI resolves the allowed dependency versions on each run.

License

MIT. See LICENSE.

mistrfilda / amqp

Package info

Security

Statistics