Queues and Background Jobs

Queues move slow or retryable work out of the request-response path. AssegaiPHP provides first-party drivers for:

• RabbitMQ through assegaiphp/rabbitmq
• Beanstalkd through assegaiphp/beanstalkd

Both drivers implement the queue contracts from assegaiphp/common. Application code produces and processes typed domain jobs; the driver owns transport decoding and delivery settlement.

Use a queue when:

• work should survive the request that created it
• failures should be retried
• traffic arrives faster than downstream work can finish
• email, exports, notifications, or integrations would otherwise delay an HTTP response

Retrying transports provide at-least-once delivery, not exactly-once delivery. A processor may receive the same logical job more than once, so make its effects idempotent.

Install a driver and the Console

Install the driver that matches the queue backend:

composer require assegaiphp/rabbitmq:^1.1.1 --with-all-dependencies

composer require assegaiphp/beanstalkd:^1.1.1 --with-all-dependencies

The RabbitMQ driver communicates through PhpAmqpLib and does not require the PECL amqp extension.

Queue discovery and worker commands come from Console:

composer global require assegaiphp/console:^0.10.3 --with-all-dependencies

Configure named queue connections

Queue configuration lives in config/queues.php. Each connection uses a driver.connection path such as rabbitmq.notifications or beanstalk.notifications.

RabbitMQ

<?php

use Assegai\Rabbitmq\RabbitMQQueue;

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

no_acknowledgement: false enables manual acknowledgement and is the safe default. The driver:

• declares exchange_name with the configured type, durability, and auto-delete policy
• binds the queue to that exchange with routing_key
• acknowledges a delivery only after decoding and processor execution succeed
• nacks a decoding or processor failure
• requeues a failed delivery when requeue_on_failure is true

Set requeue_on_failure: false when the broker should discard the delivery or route it through a configured dead-letter exchange instead of repeatedly returning a poison message to the same queue.

When exchange_name is empty, the driver publishes through RabbitMQ's default exchange using the configured routing key or queue name.

Beanstalkd

<?php

use Assegai\Beanstalkd\BeanstalkQueue;

return [
  'drivers' => [
    'beanstalk' => BeanstalkQueue::class,
  ],
  'connections' => [
    'beanstalk' => [
      'notifications' => [
        'host' => 'localhost',
        'port' => 11300,
        'connection_timeout' => 10,
        'receive_timeout' => 10,
        'reserve_timeout' => 0,
        'retry_priority' => 1024,
        'retry_delay' => 15,
      ],
    ],
  ],
];

Each poll watches only the configured tube. A named worker removes the implicit default tube from its watch list so it cannot reserve work from another queue.

reserve_timeout is the number of seconds a driver poll waits for one job. A value of 0 lets the worker observe an empty tube immediately and apply its own sleep or stopping policy.

The driver deletes a job only after the processor succeeds. A decoding or processor failure releases the job with retry_priority and retry_delay. Use a non-zero production retry delay so a repeatedly failing job does not create a tight retry loop.

Define a domain job

A queue job should describe application work without exposing transport types:

<?php

declare(strict_types=1);

namespace Assegaiphp\BlogApi\Notifications\Jobs;

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

Readonly constructor DTOs work well because their required state and types are explicit.

Produce a typed job

Import QueueInterface from Common and inject a named connection with #[InjectQueue]:

<?php

declare(strict_types=1);

namespace Assegaiphp\BlogApi\Notifications;

use Assegai\Common\Interfaces\Queues\QueueInterface;
use Assegai\Core\Attributes\Injectable;
use Assegai\Core\Queues\Attributes\InjectQueue;
use Assegaiphp\BlogApi\Notifications\Jobs\NotificationJob;

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

  public function send(string $recipient, string $message): void
  {
    $this->queue->add(new NotificationJob($recipient, $message));
  }
}

add() writes a versioned JSON envelope containing the job's class metadata and payload. This lets the consuming driver validate and rebuild the domain object before application code receives it.

Process the declared job type

A processor is an injectable provider marked with #[QueueProcessor]. Its public handler declares the concrete job type:

<?php

declare(strict_types=1);

namespace Assegaiphp\BlogApi\Notifications;

use Assegai\Core\Attributes\Injectable;
use Assegai\Core\Queues\Attributes\QueueProcessor;
use Assegaiphp\BlogApi\Notifications\Jobs\NotificationJob;

#[Injectable]
#[QueueProcessor('rabbitmq.notifications')]
final class NotificationsProcessor
{
  public function process(NotificationJob $job): void
  {
    // Send the notification using an idempotency key from the domain workflow.
  }
}

The Console passes this concrete callable directly to the driver. Common discovers the first parameter type, and the driver hydrates NotificationJob before calling process().

The handler can also be named handle, be invokable, or be selected through the processor attribute. It must be public, accept one job, and require no additional arguments.

The application processor does not decode JSON, create a process-result object, acknowledge RabbitMQ messages, or delete and release Beanstalkd jobs. Those are driver responsibilities tied to the callback outcome.

Register the processor as a provider

Processor discovery follows module provider registration:

<?php

namespace Assegaiphp\BlogApi\Notifications;

use Assegai\Core\Attributes\Modules\Module;

#[Module(
  providers: [NotificationsService::class, NotificationsProcessor::class],
)]
final class NotificationsModule
{
}

Registering the provider also lets the container resolve its constructor dependencies.

Run workers with the CLI

List configured connections and discovered processors:

assegai queue:list

Run the processor for a named connection:

assegai queue:work rabbitmq.notifications

Each driver process() call handles at most one delivery. That shared rule makes the worker options predictable:

assegai queue:work rabbitmq.notifications --once
assegai queue:work rabbitmq.notifications --max-jobs=100
assegai queue:work rabbitmq.notifications --stop-when-empty
assegai queue:work rabbitmq.notifications --sleep=1000

• --once performs one poll and exits after a delivery, failure, or empty result
• --max-jobs stops after that many jobs complete successfully
• --stop-when-empty exits when no delivery is available; it also exits after reporting a decoding or transport failure that produced no hydrated job
• --sleep sets the delay in milliseconds after an empty poll or failed delivery before the worker polls again

Use --processor=Fully\\Qualified\\ProcessorClass when more than one processor targets the same connection.

How job hydration works

Common's JSON codec preserves a job's top-level class and JSON-safe state in a versioned envelope. On delivery it:

• discovers the processor's declared job type from the callback
• validates that envelope class metadata is compatible with the declared class or interface
• hydrates constructor arguments and properties, including readonly DTOs
• restores supported nested objects, enums, dates, arrays, nullable values, and inherited private state

Legacy JSON objects without an Assegai envelope remain consumable. When the processor declares a concrete class, the codec hydrates that class from the legacy object. A processor typed only as object receives stdClass; envelope metadata does not cause an arbitrary class to be instantiated for an untyped processor.

If reflection-based JSON hydration does not match an integration, configure a custom implementation of QueueJobCodecInterface on the driver as job_codec.

Failures and settlement

Common defines one canonical process result for a single delivery attempt. The drivers populate it with callback data, the hydrated job when available, and any captured errors. Console reads that result to count successful jobs, report failures, apply worker backoff, and decide when to stop.

Every processing and settlement path catches Throwable, so both Exception and PHP Error failures follow the same controlled result and retry policy.

Settlement happens after processing:

• RabbitMQ acknowledges after success and nacks after decoding or processor failure
• Beanstalkd deletes after success and releases after decoding or processor failure

Because a process can fail after an external side effect but before settlement completes, retries can deliver the same job again. Use stable job identifiers, unique constraints, or an application-level idempotency record around non-repeatable effects.

Migrating transport-shaped processors

Keep the migration at the application boundary:

• replace a RabbitMQ processor parameter typed as AMQPMessage with a concrete domain job class
• replace a Beanstalkd processor parameter typed as a raw JSON string with the same domain job class
• replace generic transport payloads with a concrete type when the job contract is known; keep object only when receiving stdClass is intentional
• remove application-owned JSON decoding and transport settlement calls
• verify that old JSON objects contain the constructor fields required by the new job class
• make retryable processor effects idempotent before enabling automatic requeue or release behavior

Read next

• Events In Depth explains when work belongs in-process, in an outbox, or on a queue.
• Common queue contracts
• RabbitMQ driver
• Beanstalkd driver
• Console worker commands