Messaging

The messaging system provides asynchronous message processing through a bus-and-transport architecture. Messages are dispatched onto one or more transports (backed by Redis queues) and consumed by long-running worker processes. Synchronous (in-process) processing is also supported.

Core concepts

Message — a plain PHP object carrying the data for one unit of work.
Envelope — wraps a message with stamps and metadata for tracking processing state.
Transport — the queue backend. Each transport maps to one named Redis list or sorted set.
Handler — a callable (typically a class with __invoke) that processes a message.
Message bus — orchestrates dispatch through a middleware pipeline:
1. SendMessageMiddleware — routes the envelope to the configured transport(s).
2. HandleMessageMiddleware — invokes all registered handlers (runs only inside a worker, when the envelope carries a ReceivedStamp).

Defining a message

A message is a plain PHP object. For messages that represent a single entity, implement IdAwareInterface so the envelope uses the entity ID as its message ID (prevents ID collisions when the same entity is re-dispatched). Implement JsonSerializableInterface for compact Redis storage.

use Amarant\Framework\Contract\Messaging\IdAwareInterface;
use Amarant\Framework\Contract\Serializer\JsonSerializableInterface;
use Amarant\Framework\Trait\Serializer\JsonSerializeTrait;
use Amarant\Framework\Attribute\Serializer\Json;

class ProductIndexMessage implements IdAwareInterface, JsonSerializableInterface
{
    use JsonSerializeTrait;

    public function __construct(
        #[Json]
        public readonly int $id
    ) {}

    public function getId(): int|string
    {
        return $this->id;
    }
}

Dispatching a message

Inject MessageBusInterface and call dispatch() with an Envelope.

use Amarant\Framework\Contract\Messaging\MessageBusInterface;
use Amarant\Framework\Messaging\Envelope;

$this->messageBus->dispatch(Envelope::create(new ProductIndexMessage($product->getId())));

dispatch() returns the envelope with stamps added by the middleware and the transport.

Writing a handler

A handler is any class with __invoke. It receives the message object directly.

class ProductIndexMessageHandler
{
    public function __invoke(ProductIndexMessage $message): void
    {
        // process $message->id
    }
}

Bulk handlers

When a transport is configured with TransportConfigurationInterface::FLAG_BULK_MESSAGES, the worker collects all available messages and delivers them as a single BulkMessageInterface. Handlers on bulk transports must accept either form:

use Amarant\Framework\Contract\Messaging\BulkMessageInterface;

class ProductIndexMessageHandler
{
    public function __invoke(BulkMessageInterface|ProductIndexMessage $message): void
    {
        if ($message instanceof BulkMessageInterface) {
            foreach ($message->getMessages() as $single) {
                $this->process($single);
            }
            return;
        }
        $this->process($message);
    }
}

Warning

The worker cannot interrupt processing mid-bulk. If a shutdown signal arrives while a bulk handler is running, the worker exits via SIGKILL after the handler returns. Keep bulk handlers fast or process items incrementally.

Registering a message

Registration is split across two files:

Configuration/Messaging.php (or your module's DI configurator) — binds the transport instance and tags it so the messaging system can discover it.
etc/messaging.php — maps the message class to the transport ID and handler class.

1. Bind the transport

Vendor/ModuleName/Configuration/Messaging.php
use Amarant\Framework\Contract\Di\ConfigurableContainerInterface;
use Amarant\Framework\Messaging\Transport\RedisUniqueTransport;
use Amarant\Framework\Enum\DiEnum;

$container
    ->bind(
        'messaging.transport.vendor.module.product_index',       // (1)
        fn (ConfigurableContainerInterface $container) => $container->getBindBuilder()
            ->parent(RedisUniqueTransport::class)                // (2)
            ->argument('name', 'vendor_module_product_index')    // (3)
            ->build()
    )
    ->tag(
        ['messaging.transport.vendor.module.product_index'],
        [DiEnum::MESSAGING_TRANSPORT]                           // (4)
    )
    ->tag(
        [ProductIndexMessageHandler::class],
        [DiEnum::MESSAGING_HANDLER]                             // (5)
    );

Choose a unique DI binding ID. Use a dot-separated namespace that mirrors your module path.
Inherit from RedisTransport::class for a standard FIFO list, or RedisUniqueTransport::class for a sorted-set queue that deduplicates messages by message ID.
The name argument becomes the Redis key suffix: {AMA_MERCHANT_PUBLIC_ID}_{name}. Use underscores.
Tag with DiEnum::MESSAGING_TRANSPORT so MessagingLocator can discover it.
Tag the handler class with DiEnum::MESSAGING_HANDLER.

2. Map message → transport + handler

Vendor/ModuleName/etc/messaging.php
use Amarant\Framework\Contract\Application\ContextInterface;
use Amarant\Framework\Contract\Messaging\MessagingConfigurationInterface;
use Amarant\Framework\Contract\Messaging\MessagingConfiguratorInterface;
use Vendor\ModuleName\Messaging\Handler\ProductIndexMessageHandler;
use Vendor\ModuleName\Messaging\Message\ProductIndexMessage;

return new class () implements MessagingConfiguratorInterface {
    public function configure(
        ContextInterface $context,
        MessagingConfigurationInterface $messagingConfiguration
    ): void {
        $transportId = $_SERVER['MESSAGING_TRANSPORT_VENDOR_MODULE_PRODUCT_INDEX']; // (1)

        $messagingConfiguration->addTransport(ProductIndexMessage::class, $transportId);
        $messagingConfiguration->addHandler(ProductIndexMessage::class, ProductIndexMessageHandler::class);
    }
};

Always read the transport ID from an environment variable. The value must match the DI binding ID from step 1 (e.g. messaging.transport.vendor.module.product_index). This allows deployments to swap transports without code changes.

Transport types

Class	Redis structure	Deduplication	When to use
`RedisTransport`	LIST (`RPUSH`/`LPOP`)	No	General ordered work queues
`RedisUniqueTransport`	SORTED SET (`ZADD`/`ZPOPMIN`)	Yes — by message ID	When the same entity may be enqueued multiple times before processing
`SyncTransport`	None (in-process)	No	Testing or operations that must complete inline

RedisTransport and RedisUniqueTransport are both template bindings in the framework DI. Use parent(RedisTransport::class) or parent(RedisUniqueTransport::class) when binding your transport — this inherits the Redis client and the key prefix from the framework configuration automatically.

The key prefix is taken from the AMA_MERCHANT_PUBLIC_ID environment variable.

Transport configuration

After calling addTransport(), retrieve the transport's configuration object to set additional options.

use Amarant\Framework\Contract\Messaging\TransportConfigurationInterface;

$config = $messagingConfiguration->getTransportConfiguration($transportId);

// Deliver all available messages as a single BulkMessage to the handler
$config->flag(TransportConfigurationInterface::FLAG_BULK_MESSAGES, true);

// Retry a failed message up to N times before dead-lettering it
$config->flag(TransportConfigurationInterface::ALLOWED_RETRY_COUNT, 3);

Constant	Type	Default	Description
`FLAG_BULK_MESSAGES`	bool	`false`	Collect all available messages into a `BulkMessage` before invoking the handler.
`ALLOWED_RETRY_COUNT`	int	`0`	How many times to retry a message whose handler throws a non-fatal exception before dead-lettering it.
`FLAG_AUTO_START`	bool	`false`	Hint for process managers to auto-start a worker for this transport.

Note

Retries are immediate — there is no back-off delay between attempts. If your handler requires time-delayed redelivery, implement that logic yourself (e.g. re-dispatch with a future-dated payload).

Unrecoverable errors

Throw Amarant\Framework\Exception\UnrecoverableMessagingException from a handler to signal that the message must not be retried regardless of ALLOWED_RETRY_COUNT. The worker dead-letters the message and stops processing that transport.

Workers

Two worker types are available:

Class	Implementation	When to use
`Worker`	Single in-process consumer	Development, debugging, or single-threaded workloads
`ForkedMultiWorker`	Forks N child processes via `spatie/fork`	Production — default `MultiWorkerInterface` binding
`MultiWorker`	Spawns N child processes via Symfony Process	Higher isolation; each child is a fully independent PHP process

ForkedMultiWorker is the default implementation resolved for MultiWorkerInterface.

Stop conditions

A worker stops cleanly when any of these conditions is met:

Maximum run time exceeded (configurable; defaults to 24 hours).
Available memory exceeded.
Consecutive error budget exhausted.
MessagingStateInterface::isStopped() returns true (Redis-based graceful shutdown).
An UnrecoverableMessagingException is caught.
exitIfNoMessage is set and the queue is empty.

Running via CLI

For development or one-off drains:

# Single worker — one or more transport identifiers
php bin/ama messaging:worker:run messaging.transport.vendor.module.product_index

# Multiple transports on one worker
php bin/ama messaging:worker:run messaging.transport.a messaging.transport.b

# Exit as soon as the queue is empty (useful in cron jobs)
php bin/ama messaging:worker:run messaging.transport.vendor.module.product_index --exit-if-no-messages=1

# Multi-worker — spawn N parallel workers on the same transport
php bin/ama messaging:multi-worker:run messaging.transport.vendor.module.product_index --instances=4 --install-signal-handler

Available options for both commands:

Option	Description
`--max-messages=N`	Stop after processing N messages.
`--max-runtime=N`	Stop after N seconds.
`--max-memory=N`	Stop when memory usage reaches N bytes.
`--max-errors=N`	Stop after N consecutive errors.
`--exit-if-no-messages=1`	Stop when the queue is empty instead of polling.
`--restart-container=1`	Reset the DI container between messages (reduces memory growth for long-running workers).
`--instances=N`	(multi-worker only) Number of parallel worker processes.
`--install-signal-handler`	(multi-worker only) Install SIGINT/SIGTERM handler to stop the pool gracefully.

Programmatic usage (Jobs)

The recommended way to run workers in production is inside a scheduled Job. Inject MultiWorkerInterface, call factory() to get a fresh instance, install signal handlers, and call run().

use Amarant\Framework\Contract\Messaging\MultiWorkerInterface;
use Amarant\Framework\Contract\Messaging\SignalAwareWorkerInterface;
use Amarant\Framework\Contract\Scheduling\JobContextInterface;
use Amarant\Framework\Contract\Scheduling\JobInterface;
use Amarant\Framework\Contract\Scheduling\JobConfigurationInterface;
use Amarant\Framework\Messaging\Worker\MultiWorkerConfiguration;
use Amarant\Framework\Messaging\Worker\WorkerConfiguration;
use Psr\Log\LoggerAwareInterface;

final class HandleProductIndexMessagesJob implements JobInterface
{
    public function __construct(
        private readonly MultiWorkerInterface $worker
    ) {}

    public static function getName(): string
    {
        return 'vendor_module_handle_product_index_messages';
    }

    public function execute(JobContextInterface $context): void
    {
        $worker = $this->worker->factory();

        if ($worker instanceof LoggerAwareInterface) {
            $worker->setLogger($context->getLogger());
        }

        if ($worker instanceof SignalAwareWorkerInterface && $worker->shouldInstallSignalHandler()) {
            $sigHandler = function () use ($worker) {
                $worker->stop();
                return true;
            };
            \pcntl_signal(SIGINT, $sigHandler);
            \pcntl_signal(SIGTERM, $sigHandler);
        }

        $worker->run(
            configurations: [
                new MultiWorkerConfiguration(
                    instances: 4,                                          // (1)
                    configuration: new WorkerConfiguration(
                        transportIdentifiers: [
                            'messaging.transport.vendor.module.product_index'
                        ],
                        maxRunTime: 55,                                    // (2)
                    )
                ),
            ],
            exitIfNoMessage: true                                          // (3)
        );
    }

    public function configure(JobConfigurationInterface $configuration): void
    {
        $configuration
            ->lock(true)
            ->everyMinute();
    }
}

Number of parallel worker processes to spawn.
Set maxRunTime slightly under the job schedule interval to ensure workers exit before the next run.
Exit as soon as the queue is empty rather than waiting for maxRunTime.

Graceful shutdown

Inject MessagingStateInterface to remotely signal all workers to finish their current message and exit.

$messagingState->stop();      // sets a Redis flag; workers drain current message and exit
$messagingState->start();     // clears the flag (ready to run again)
$messagingState->isStopped(); // check current state

Environment variables

Variable	Description
`MESSAGING_REDIS_URI`	Predis connection URI for all transports (e.g. `tcp://127.0.0.1:6379`).
`MESSAGING_STATE_REDIS_URI`	Predis connection URI for the graceful-shutdown state key.
`AMA_MERCHANT_PUBLIC_ID`	Prefix for all Redis transport keys.
`MESSAGING_TRANSPORT_*`	One variable per transport; value is the DI binding ID registered in `Configuration/Messaging.php`.