Nette
This page adds Hermes to a fresh nette/web-project. A presenter emits user.registered, and php bin/console hermes:work handles it. Read Framework integration first for the parts every framework shares.
1. Install
composer create-project nette/web-project app
cd app
composer require tomaj/hermes:dev-master contributte/console
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. Nette has no console of its own; contributte/console adds Symfony Console commands to the DI container.
2. Configure the services
All of Hermes lives in one NEON file. Put the driver and the queue name for each environment in config/local.neon, and give each application its own queue.
extensions:
console: Contributte\Console\DI\ConsoleExtension(%consoleMode%)
parameters:
consoleMode: false
hermes:
driver: redis # redis (ext-redis), predis, rabbitmq or sqs; override it in local.neon
queue: hermes
max_items: 1000
redis:
host: 127.0.0.1
port: 6379
rabbitmq:
host: 127.0.0.1
port: 5672
user: guest
password: guest
vhost: /
sqs:
region: eu-central-1
endpoint: null # http://127.0.0.1:4566 for a local SQS
key: ''
secret: ''
services:
# One driver for the web process and the worker.
hermes.driver:
type: Tomaj\Hermes\Driver\DriverInterface
create: App\Hermes\DriverFactory::create(%hermes%)
- Tomaj\Hermes\Shutdown\SharedFileShutdown(%tempDir%/hermes-shutdown)
# Hermes logs through Tracy's logger (log/info.log).
- Tracy\Bridges\Psr\TracyToPsrLoggerAdapter
# Producers type-hint EmitterInterface.
- Tomaj\Hermes\Emitter
# The worker command type-hints DispatcherInterface.
hermes.dispatcher:
create: Tomaj\Hermes\Dispatcher
setup:
- registerHandler('user.registered', @App\Hermes\UserRegisteredHandler)
- App\Hermes\UserRegisteredHandler
- App\Console\HermesWorkCommand
- App\Console\HermesRestartCommand
- App\Console\RegisterUserCommandDriverFactory turns the hermes parameter into a driver:
<?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);
}
}Bootstrap loads hermes.neon and gets a bootConsoleApplication() for bin/console:
<?php
declare(strict_types=1);
namespace App;
use Nette;
use Nette\Bootstrap\Configurator;
class Bootstrap
{
private readonly Configurator $configurator;
private readonly string $rootDir;
public function __construct()
{
$this->rootDir = dirname(__DIR__);
$this->configurator = new Configurator();
$this->configurator->setTempDirectory($this->rootDir . '/temp');
}
public function bootWebApplication(): Nette\DI\Container
{
$this->initializeEnvironment();
$this->setupContainer();
return $this->configurator->createContainer();
}
public function bootConsoleApplication(): Nette\DI\Container
{
$this->configurator->addStaticParameters(['consoleMode' => true]);
$this->initializeEnvironment();
$this->setupContainer();
return $this->configurator->createContainer();
}
public function initializeEnvironment(): void
{
//$this->configurator->setDebugMode('secret@23.75.345.200'); // enable for your remote IP
$this->configurator->enableTracy($this->rootDir . '/log');
$this->configurator->createRobotLoader()
->addDirectory(__DIR__)
->register();
}
private function setupContainer(): void
{
$configDir = $this->rootDir . '/config';
$this->configurator->addConfig($configDir . '/common.neon');
$this->configurator->addConfig($configDir . '/services.neon');
$this->configurator->addConfig($configDir . '/hermes.neon');
if (is_file($configDir . '/local.neon')) {
$this->configurator->addConfig($configDir . '/local.neon'); // per-machine overrides, not in git
}
}
}#!/usr/bin/env php
<?php
declare(strict_types=1);
require __DIR__ . '/../vendor/autoload.php';
$bootstrap = new App\Bootstrap();
$container = $bootstrap->bootConsoleApplication();
$application = $container->getByType(Contributte\Console\Application::class);
exit($application->run());3. Log through Tracy
TracyToPsrLoggerAdapter gives Hermes Tracy's logger as LoggerInterface. The Start handle message lines go to log/info.log. Tracy has no notice level, so the notice Exiting hermes dispatcher - shutdown lands in log/warning.log. Register a Monolog logger as LoggerInterface instead if you want PSR-3 levels as they are. See Logging.
4. Emit from a presenter
<?php
declare(strict_types=1);
namespace App\Presentation\User;
use Nette\Application\Attributes\Requires;
use Nette\Application\UI\Presenter;
use Tomaj\Hermes\EmitterInterface;
use Tomaj\Hermes\Message;
final class UserPresenter extends Presenter
{
public function __construct(private readonly EmitterInterface $emitter)
{
parent::__construct();
}
/** POST /user/create with an email field */
#[Requires(methods: 'POST')]
public function actionCreate(): void
{
$email = $this->getHttpRequest()->getPost('email');
if (!is_string($email) || filter_var($email, FILTER_VALIDATE_EMAIL) === false) {
$this->getHttpResponse()->setCode(422);
$this->sendJson(['error' => 'A valid email is required.']);
}
// Save the user here, then hand the slow work to the worker.
$this->emitter->emit(new Message('user.registered', ['email' => $email]));
$this->getHttpResponse()->setCode(201);
$this->sendJson(['status' => 'registered']);
}
}POST /user/create with email returns 201, a missing email 422, and GET 405. The producer command, used below:
<?php
declare(strict_types=1);
namespace App\Console;
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;
}
}5. Write a handler
<?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;
}
}Throw to retry, return false to give up; see Retry.
6. Run the worker
<?php
declare(strict_types=1);
namespace App\Console;
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;
}
}php bin/console hermes:work # terminal 1
php bin/console app:register-user jane@example.com # terminal 2
grep 'Welcome email sent' log/info.logAfter you change a NEON file, delete temp/cache if the container does not rebuild (it rebuilds on its own in debug mode).
7. Restart workers on deploy
<?php
declare(strict_types=1);
namespace App\Console;
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 php bin/console hermes:restart as the last deploy step, after the cache is cleared. Workers stop after the current message, and Supervisor starts new ones:
; /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.logThe shutdown file is temp/hermes-shutdown. If a deploy replaces the temp folder, workers in the old release never see the new file; use a path outside the release, such as a shared folder. See Shutdown and max items and Production deployment.
8. Test the producer
Pass a mock of EmitterInterface to the command. From Hermes 6.0 Emitter is final and cannot be mocked.
<?php
declare(strict_types=1);
namespace Tests;
use App\Console\RegisterUserCommand;
use PHPUnit\Framework\TestCase;
use Symfony\Component\Console\Tester\CommandTester;
use Tomaj\Hermes\EmitterInterface;
use Tomaj\Hermes\MessageInterface;
final class RegisterUserCommandTest extends TestCase
{
public function testEmitsUserRegistered(): void
{
// 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();
$tester = new CommandTester(new RegisterUserCommand($emitter));
self::assertSame(0, $tester->execute(['email' => 'jane@example.com']));
}
}Retry jitter and a Random\Randomizer service
From Hermes 6.0, Dispatcher takes an optional ?Random\Randomizer $randomizer (see Framework integration). Nette autowires it: if the container has a Random\Randomizer service, the dispatcher gets it. To keep the default secure engine, on Hermes 6.0 or newer write the dispatcher like this:
hermes.dispatcher:
create: Tomaj\Hermes\Dispatcher(randomizer: null)On Hermes 5.x Nette fails with Unable to pass specified arguments to Tomaj\Hermes\Dispatcher::__construct(), so leave it out there.
Full source
Every file, with install.sh, e2e.sh and the Supervisor program, is in examples/frameworks/nette on GitHub. Verified with nette/application 3.3.0, nette/di 3.2.6, contributte/console 0.11.0 and Tracy 2.12 on PHP 8.4, with every driver.
FAQ
Why contributte/console and not a plain PHP script?
The commands get their dependencies from the container, like presenters do, and you get --priority and help for free. A plain bin/worker.php that builds the container and calls handle() works too; see the welcome email tutorial.
Can I use Monolog instead of Tracy?
Yes. Register a Monolog\Logger service and remove the TracyToPsrLoggerAdapter line, so only one LoggerInterface is autowired. See Logging.
Where do I put the driver per environment?
In config/local.neon, which Bootstrap loads last. For example parameters: hermes: driver: rabbitmq. See Drivers overview to choose one.
Does the presenter wait for the handler?
No. emit() only writes the message to the queue and returns; the worker handles it later. See Concepts.
How do I run more workers?
Raise numprocs in the Supervisor program. Each worker takes one message at a time from the same queue. See Production deployment.