Symfony
This page adds Hermes to a fresh Symfony 7.4 project, next to or instead of Messenger. A controller emits user.registered, and bin/console hermes:work handles it. Read Framework integration first for the parts every framework shares.
1. Install
bash
composer create-project symfony/skeleton:"7.4.*" app
cd app
composer require tomaj/hermes:dev-master symfony/monolog-bundle
composer require predis/predis # only for the predis driver
composer require php-amqplib/php-amqplib # only for the rabbitmq driver
composer require aws/aws-sdk-php # only for the sqs driverUntil Hermes 6.0 is tagged, require tomaj/hermes:dev-master as above; switch to ^6.0 once it is released. Without monolog-bundle the autowired LoggerInterface is Symfony's minimal logger, which writes to stderr.
2. Configure the services
The whole config/services.yaml is below. The hermes parameter collects the env vars, the DriverFactory builds the driver from it, and the two interfaces are aliases for the Hermes classes. The handler is registered with a calls entry on the dispatcher.
yaml
# yaml-language-server: $schema=../vendor/symfony/dependency-injection/Loader/schema/services.schema.json
parameters:
# Defaults for the environment variables; override them in .env.local or the real environment.
env(HERMES_DRIVER): redis # redis (ext-redis), predis, rabbitmq or sqs
env(HERMES_QUEUE): hermes
env(HERMES_REDIS_HOST): 127.0.0.1
env(HERMES_REDIS_PORT): '6379'
env(HERMES_RABBITMQ_HOST): 127.0.0.1
env(HERMES_RABBITMQ_PORT): '5672'
env(HERMES_RABBITMQ_USER): guest
env(HERMES_RABBITMQ_PASSWORD): guest
env(HERMES_RABBITMQ_VHOST): /
env(HERMES_SQS_REGION): eu-central-1
env(HERMES_SQS_ENDPOINT): ''
env(AWS_ACCESS_KEY_ID): ''
env(AWS_SECRET_ACCESS_KEY): ''
hermes:
driver: '%env(HERMES_DRIVER)%'
queue: '%env(HERMES_QUEUE)%'
max_items: 1000
redis:
host: '%env(HERMES_REDIS_HOST)%'
port: '%env(int:HERMES_REDIS_PORT)%'
rabbitmq:
host: '%env(HERMES_RABBITMQ_HOST)%'
port: '%env(int:HERMES_RABBITMQ_PORT)%'
user: '%env(HERMES_RABBITMQ_USER)%'
password: '%env(HERMES_RABBITMQ_PASSWORD)%'
vhost: '%env(HERMES_RABBITMQ_VHOST)%'
sqs:
region: '%env(HERMES_SQS_REGION)%'
endpoint: '%env(HERMES_SQS_ENDPOINT)%'
key: '%env(AWS_ACCESS_KEY_ID)%'
secret: '%env(AWS_SECRET_ACCESS_KEY)%'
services:
_defaults:
autowire: true
autoconfigure: true
App\:
resource: '../src/'
# One driver for the web process and the worker.
Tomaj\Hermes\Driver\DriverInterface:
factory: ['App\Hermes\DriverFactory', 'create']
arguments: ['%hermes%']
Tomaj\Hermes\Shutdown\ShutdownInterface:
class: Tomaj\Hermes\Shutdown\SharedFileShutdown
arguments: ['%kernel.project_dir%/var/hermes-shutdown']
# Producers type-hint EmitterInterface; the logger is autowired (Monolog).
Tomaj\Hermes\EmitterInterface:
class: Tomaj\Hermes\Emitter
# The worker command type-hints DispatcherInterface.
Tomaj\Hermes\DispatcherInterface:
class: Tomaj\Hermes\Dispatcher
calls:
- registerHandler: ['user.registered', '@App\Hermes\UserRegisteredHandler']Set HERMES_DRIVER to redis, predis, rabbitmq or sqs in .env.local, and give each application its own HERMES_QUEUE.
php
<?php
declare(strict_types=1);
namespace App\Hermes;
use Aws\Sqs\SqsClient;
use InvalidArgumentException;
use PhpAmqpLib\Connection\AMQPLazyConnection;
use Predis\Client;
use Redis;
use Tomaj\Hermes\Driver\AmazonSqsDriver;
use Tomaj\Hermes\Driver\DriverInterface;
use Tomaj\Hermes\Driver\LazyRabbitMqDriver;
use Tomaj\Hermes\Driver\PredisSetDriver;
use Tomaj\Hermes\Driver\RedisSetDriver;
/**
* Builds the Hermes driver from plain configuration, so the web process and the worker
* always use the same driver, queue name and connection.
*/
final class DriverFactory
{
/**
* @param array{
* driver: string,
* queue: string,
* max_items: int,
* redis: array{host: string, port: int},
* rabbitmq: array{host: string, port: int, user: string, password: string, vhost: string},
* sqs: array{region: string, endpoint: ?string, key: string, secret: string},
* } $config
*/
public static function create(array $config): DriverInterface
{
$queue = $config['queue'];
$driver = match ($config['driver']) {
'redis' => new RedisSetDriver(self::redis($config['redis']), $queue),
'predis' => new PredisSetDriver(new Client(['host' => $config['redis']['host'], 'port' => $config['redis']['port']]), $queue),
'rabbitmq' => new LazyRabbitMqDriver(
new AMQPLazyConnection(
$config['rabbitmq']['host'],
$config['rabbitmq']['port'],
$config['rabbitmq']['user'],
$config['rabbitmq']['password'],
$config['rabbitmq']['vhost'],
),
$queue,
[],
1, // seconds to sleep when the queue is empty
),
'sqs' => new AmazonSqsDriver(self::sqs($config['sqs']), $queue),
default => throw new InvalidArgumentException("Unknown Hermes driver \"{$config['driver']}\""),
};
// Start a fresh worker process after this many messages.
$driver->setMaxProcessItems($config['max_items']);
return $driver;
}
/**
* @param array{host: string, port: int} $config
*/
private static function redis(array $config): Redis
{
$redis = new Redis();
$redis->connect($config['host'], $config['port']);
return $redis;
}
/**
* @param array{region: string, endpoint: ?string, key: string, secret: string} $config
*/
private static function sqs(array $config): SqsClient
{
$options = [
'region' => $config['region'],
'version' => 'latest',
'credentials' => ['key' => $config['key'], 'secret' => $config['secret']],
];
if ($config['endpoint'] !== null && $config['endpoint'] !== '') {
$options['endpoint'] = $config['endpoint']; // a local SQS such as moto or ElasticMQ
}
return new SqsClient($options);
}
}Monolog is autowired as LoggerInterface, so the Hermes lines land in var/log/dev.log (or your production handlers). See Logging.
3. Emit from a controller
php
<?php
declare(strict_types=1);
namespace App\Controller;
use Symfony\Component\HttpFoundation\JsonResponse;
use Symfony\Component\HttpFoundation\Request;
use Symfony\Component\Routing\Attribute\Route;
use Tomaj\Hermes\EmitterInterface;
use Tomaj\Hermes\Message;
final class UserController
{
public function __construct(private readonly EmitterInterface $emitter) {}
#[Route('/users', name: 'user_register', methods: ['POST'])]
public function register(Request $request): JsonResponse
{
$email = $request->getPayload()->getString('email');
if (filter_var($email, FILTER_VALIDATE_EMAIL) === false) {
return new JsonResponse(['error' => 'A valid email is required.'], 422);
}
// Save the user here, then hand the slow work to the worker.
$this->emitter->emit(new Message('user.registered', ['email' => $email]));
return new JsonResponse(['status' => 'registered'], 201);
}
}The producer command, used below to try it:
php
<?php
declare(strict_types=1);
namespace App\Command;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Tomaj\Hermes\EmitterInterface;
use Tomaj\Hermes\Message;
#[AsCommand(name: 'app:register-user', description: 'Emit user.registered for one email address')]
final class RegisterUserCommand extends Command
{
public function __construct(private readonly EmitterInterface $emitter)
{
parent::__construct();
}
#[\Override]
protected function configure(): void
{
$this->addArgument('email', InputArgument::REQUIRED);
}
#[\Override]
protected function execute(InputInterface $input, OutputInterface $output): int
{
$email = (string) $input->getArgument('email');
$this->emitter->emit(new Message('user.registered', ['email' => $email]));
$output->writeln("Queued the welcome email for {$email}.");
return Command::SUCCESS;
}
}4. Write a handler
php
<?php
declare(strict_types=1);
namespace App\Hermes;
use Psr\Log\LoggerInterface;
use Tomaj\Hermes\Handler\HandlerInterface;
use Tomaj\Hermes\Handler\RetryTrait;
use Tomaj\Hermes\MessageInterface;
final class UserRegisteredHandler implements HandlerInterface
{
use RetryTrait;
public function __construct(private readonly LoggerInterface $logger) {}
#[\Override]
public function handle(MessageInterface $message): bool
{
$email = $message->getPayload()['email'] ?? null;
if (!is_string($email)) {
return false; // a broken message stays broken, do not retry it
}
// Send the welcome email here. Throw an exception to retry later.
$this->logger->info("Welcome email sent to {$email}", ['message_id' => $message->getId()]);
return true;
}
}It is an autowired service, so its constructor can ask for anything in the container. Throw to retry, return false to give up; see Retry.
5. Run the worker
php
<?php
declare(strict_types=1);
namespace App\Command;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Tomaj\Hermes\DispatcherInterface;
#[AsCommand(name: 'hermes:work', description: 'Run the Hermes worker until it is told to stop')]
final class HermesWorkCommand extends Command
{
public function __construct(private readonly DispatcherInterface $dispatcher)
{
parent::__construct();
}
#[\Override]
protected function configure(): void
{
$this->addOption('priority', null, InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY, 'Handle only these priorities');
}
#[\Override]
protected function execute(InputInterface $input, OutputInterface $output): int
{
/** @var list<string> $priorities */
$priorities = $input->getOption('priority');
// Blocks until shutdown (hermes:restart) or the max items limit.
$this->dispatcher->handle(array_map(intval(...), $priorities));
return Command::SUCCESS;
}
}bash
bin/console hermes:work # terminal 1
bin/console app:register-user jane@example.com # terminal 2
grep 'Welcome email sent' var/log/dev.log--priority=200 limits the worker to one priority queue; see Priority.
6. Restart workers on deploy
php
<?php
declare(strict_types=1);
namespace App\Command;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Tomaj\Hermes\Shutdown\ShutdownInterface;
#[AsCommand(name: 'hermes:restart', description: 'Stop every running Hermes worker after its current message')]
final class HermesRestartCommand extends Command
{
public function __construct(private readonly ShutdownInterface $shutdown)
{
parent::__construct();
}
#[\Override]
protected function execute(InputInterface $input, OutputInterface $output): int
{
$this->shutdown->shutdown();
$output->writeln('Workers will stop after their current message. Supervisor or systemd starts new ones.');
return Command::SUCCESS;
}
}Run bin/console hermes:restart after cache:clear. Workers stop after the current message, and Supervisor starts new ones:
ini
; /etc/supervisor/conf.d/hermes-worker.conf
; Two workers. hermes:restart stops them after their current message, and Supervisor starts new ones.
; With ext-pcntl, SIGTERM (stopsignal) also lets a worker finish its message; stopwaitsecs is the limit before SIGKILL.
[program:hermes-worker]
command=php bin/console hermes:work
directory=/var/www/app
process_name=%(program_name)s_%(process_num)02d
numprocs=2
user=www-data
autostart=true
autorestart=true
startsecs=1
stopwaitsecs=60
stopsignal=TERM
redirect_stderr=true
stdout_logfile=/var/log/hermes-worker.logSee Production deployment for systemd.
7. Test the producer
Replace EmitterInterface in the test container with a mock. From Hermes 6.0 Emitter is final and cannot be mocked.
php
<?php
declare(strict_types=1);
namespace App\Tests\Controller;
use Symfony\Bundle\FrameworkBundle\Test\WebTestCase;
use Tomaj\Hermes\EmitterInterface;
use Tomaj\Hermes\MessageInterface;
final class UserControllerTest extends WebTestCase
{
public function testRegistrationEmitsUserRegistered(): void
{
$client = static::createClient();
// Mock the interface: Emitter is final from Hermes 6.0.
$emitter = $this->createMock(EmitterInterface::class);
$emitter->expects(self::once())
->method('emit')
->with(self::callback(
static fn(MessageInterface $message): bool => $message->getType() === 'user.registered'
&& $message->getPayload() === ['email' => 'jane@example.com'],
))
->willReturnSelf();
static::getContainer()->set(EmitterInterface::class, $emitter);
$client->jsonRequest('POST', '/users', ['email' => 'jane@example.com']);
self::assertResponseStatusCodeSame(201);
}
public function testInvalidEmailEmitsNothing(): void
{
$client = static::createClient();
$emitter = $this->createMock(EmitterInterface::class);
$emitter->expects(self::never())->method('emit');
static::getContainer()->set(EmitterInterface::class, $emitter);
$client->jsonRequest('POST', '/users', ['email' => 'not-an-email']);
self::assertResponseStatusCodeSame(422);
}
}Retry jitter and a Random\Randomizer service
From Hermes 6.0, Dispatcher takes an optional ?Random\Randomizer $randomizer (see Framework integration). If your app registers a Random\Randomizer service, autowiring passes it to the dispatcher, and retry jitter comes from that engine. To keep the default secure engine, add this to the dispatcher service on Hermes 6.0 or newer:
yaml
Tomaj\Hermes\DispatcherInterface:
class: Tomaj\Hermes\Dispatcher
arguments:
$randomizer: nullOn Hermes 5.x the container fails with has no argument named "$randomizer", so leave it out there.
Full source
Every file, with install.sh, e2e.sh and the Supervisor program, is in examples/frameworks/symfony on GitHub. Verified with Symfony 7.4.21 and monolog-bundle 4.1.0 on PHP 8.4, with every driver.
FAQ
Do I need Symfony Messenger?
No. This setup does not install or use Messenger. Both can run in one app if they use different queues; see Framework integration.
Why is the handler registered in services.yaml and not with a tag?
One calls line per message type is easy to read and needs no compiler pass. With many handlers, a tagged iterator and a small loop in a factory work too. See Concepts.
The worker does not see my new code after a deploy. Why?
It is a long-running process with the old container in memory. Run bin/console hermes:restart. See Shutdown and max items.
Can the worker run with APP_ENV=prod?
Yes, and it should. Warm the cache before you start it, so that the first worker does not build the container while others wait. See Production deployment.
How do I point the sqs driver at LocalStack or moto?
Set HERMES_SQS_ENDPOINT=http://127.0.0.1:4566 and dummy AWS keys. See Amazon SQS.