README

RabbitMQ queue support for AssegaiPHP applications.

Description

This package integrates RabbitMQ with AssegaiPHP through PhpAmqpLib. It serializes queued domain jobs, hydrates them for typed processors, and settles deliveries according to the processor outcome.

Contribution workflow

For commit and pull request conventions in this repo, see:

docs/commit-and-pr-guidelines.md

Installation

Install the package with Composer:

$ composer require assegaiphp/rabbitmq

Compatibility

RabbitMQ package	AssegaiPHP Common
`>=1.1.2 <2.0`	`^0.10.1`
`1.1.1`	`^0.10.1`
`1.1.0`	`^0.10.0`
`1.0.x`	`^0.9.0`

Upgrade this package and its coordinated first-party dependencies together when moving between AssegaiPHP release lines.

Configuration

<?php

use Assegai\Rabbitmq\RabbitMQQueue;

return [
  'drivers' => [
    'rabbitmq' => RabbitMQQueue::class,
  ],
  'connections' => [
    'rabbitmq' => [
      'notifications' => [
        'host' => 'localhost',
        'port' => 5672,
        'username' => 'guest',
        'password' => 'guest',
        'vhost' => '/',
        'exchange_name' => 'notifications',
        'exchange_type' => 'direct',
        'exchange_durable' => true,
        'exchange_auto_delete' => false,
        'routing_key' => 'notifications',
        'passive' => false,
        'durable' => true,
        'exclusive' => false,
        'auto_delete' => false,
        'no_acknowledgement' => false,
        'requeue_on_failure' => true,
      ],
    ],
  ],
];

Queue references use the driver.connection format, such as rabbitmq.notifications.

Creating or injecting a queue is configuration-only. The driver opens the AMQP connection, declares the queue, and configures its exchange binding on the first broker operation. This allows an HTTP application to boot while RabbitMQ is temporarily unavailable; the operation that needs RabbitMQ receives the connection error and a later operation can retry.

Manual acknowledgement is the safe default. A successful processor call acknowledges the delivery. A decoding or processor failure nacks it and requeues it unless requeue_on_failure is false.

When exchange_name is non-empty, the driver declares that exchange and binds the queue using routing_key. With an empty exchange name, it publishes directly to the queue name.

Producing jobs

Inject a configured queue using #[InjectQueue] and add a domain job:

<?php

use Assegai\Common\Interfaces\Queues\QueueInterface;
use Assegai\Core\Attributes\Injectable;
use Assegai\Core\Queues\Attributes\InjectQueue;

final readonly class NotificationJob
{
  public function __construct(
    public string $recipient,
    public string $message,
  ) {
  }
}

#[Injectable]
readonly class NotificationsService
{
  public function __construct(
    #[InjectQueue('rabbitmq.notifications')] private QueueInterface $queue,
  ) {
  }

  public function send(NotificationJob $job): void
  {
    $this->queue->add($job);
  }
}

The driver writes a versioned JSON envelope containing the job class and payload.

Consuming jobs

Define an injectable processor whose method declares the job type it accepts:

<?php

use Assegai\Core\Attributes\Injectable;
use Assegai\Core\Queues\Attributes\QueueProcessor;

#[Injectable]
#[QueueProcessor('rabbitmq.notifications')]
final class NotificationsProcessor
{
  public function process(NotificationJob $job): void
  {
    // Handle the notification.
  }
}

Register the processor in its module's provider list so the Console can discover it. The worker validates the envelope class against the processor parameter and hydrates the domain object before invocation. Legacy JSON messages are hydrated when the processor declares a concrete class; a processor typed only as object receives stdClass.

Generate a processor with the Console when you want a starter class:

$ assegai g qp notifications --queue=rabbitmq.notifications
$ assegai g qp notifications --queue=rabbitmq.notifications --job=Jobs/NotificationJob

Running workers

Discover and run queue processors with the Assegai Console:

$ assegai queue:list
$ assegai queue:work rabbitmq.notifications

Process at most one available job and exit with --once:

$ assegai queue:work rabbitmq.notifications --once

If multiple processors target the same queue, use --processor to select one. See the AssegaiPHP queue guide for application-level worker guidance.

Testing

Run the package test suite with Composer:

$ composer test

Resources

Support

Assegai is an MIT-licensed open source project. It can grow thanks to sponsors and support by the amazing backers. If you'd like to join them, please read more here.

Stay in touch

Author - Andrew Masiye
Website - https://assegaiphp.com
Twitter - @assegaiphp

License

Assegai is MIT licensed.

assegaiphp / rabbitmq

Maintainers

Package info

Statistics

Security