Nette

Táto stránka pridá Hermes do nového nette/web-project. Presenter vloží user.registered a php bin/console hermes:work ju spracuje. Najprv si prečítajte Integráciu do frameworkov, kde sú časti spoločné pre všetky frameworky.

1. Inštalácia

bash

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 driver

Kým Hermes 6.0 nemá tag, vyžadujte tomaj/hermes:dev-master ako vyššie; po vydaní prejdite na ^6.0. Nette nemá vlastnú konzolu; contributte/console pridá do DI kontajnera príkazy Symfony Console.

2. Konfigurácia služieb

Celý Hermes je v jednom súbore NEON. Driver a názov fronty pre každé prostredie dajte do config/local.neon a každej aplikácii dajte vlastnú queue.

yaml

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\RegisterUserCommand

DriverFactory premení parameter hermes na driver:

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);
    }
}

Bootstrap načíta hermes.neon a dostane bootConsoleApplication() pre bin/console:

php

<?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
        }
    }
}

php

#!/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. Logovanie cez Tracy

TracyToPsrLoggerAdapter dá Hermesu logger Tracy ako LoggerInterface. Riadky Start handle message idú do log/info.log. Tracy nemá úroveň notice, takže oznámenie Exiting hermes dispatcher - shutdown skončí v log/warning.log. Ak chcete PSR-3 úrovne bez zmeny, zaregistrujte namiesto toho ako LoggerInterface logger Monolog. Pozrite Logovanie.

4. Vloženie správy z presentera

php

<?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 s email vráti 201, chýbajúci e-mail 422 a GET 405. Príkaz producenta, ktorý sa používa nižšie:

php

<?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. Napíšte handler

php

<?php

declare(strict_types=1);

namespace App\Hermes;

use Psr\Log\LoggerInterface;
use Tomaj\Hermes\Handler\RetryableHandlerInterface;
use Tomaj\Hermes\Handler\RetryTrait;
use Tomaj\Hermes\MessageInterface;

final class UserRegisteredHandler implements RetryableHandlerInterface
{
    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;
    }
}

Výnimkou spustíte opakovanie, vrátením false to vzdáte; pozrite Opakovanie.

6. Spustite workera

php

<?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;
    }
}

bash

php bin/console hermes:work                          # terminal 1
php bin/console app:register-user jane@example.com   # terminal 2
grep 'Welcome email sent' log/info.log

Ak sa po zmene súboru NEON kontajner neprestaví, zmažte temp/cache (v debug režime sa prestavuje sám).

7. Reštart workerov pri nasadení

php

<?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;
    }
}

Ako posledný krok nasadenia, po vymazaní cache, spustite php bin/console hermes:restart. Workery sa zastavia po aktuálnej správe a Supervisor spustí nové:

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.log

Súbor na ukončenie je temp/hermes-shutdown. Ak nasadenie nahrádza priečinok temp, workery zo starého vydania nový súbor nikdy neuvidia; použite cestu mimo vydania, napríklad zdieľaný priečinok. Pozrite Ukončenie a limit správ a Nasadenie do produkcie.

8. Otestujte producenta

Príkazu odovzdajte mock EmitterInterface. Od Hermesu 6.0 je Emitter final a nedá sa zamockovať.

php

<?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']));
    }
}

Emitovanie po commite do databázy

Ak presenter vloží riadok v transakcii a emitne o ňom správu, obaľte emitter do BufferedEmitter, aby worker nikdy nehľadal riadok, ktorý ešte nie je commitnutý. Riadok - Tomaj\Hermes\Emitter v hermes.neon nahraďte týmto, kde databaseTransaction je vaša služba s isInTransaction(), registerOnCommit() a registerOnRollback():

yaml

	hermes.emitter:
		create: Tomaj\Hermes\BufferedEmitter(Tomaj\Hermes\Emitter(@hermes.driver, @Psr\Log\LoggerInterface), [@databaseTransaction, isInTransaction])
		setup:
			- @databaseTransaction::registerOnCommit([@self, flush])
			- @databaseTransaction::registerOnRollback([@self, discard])

@self je BufferedEmitter, takže hooky odosielajú a zahadzujú tú istú inštanciu, ktorú producenti dostanú pre EmitterInterface. Overené na skompilovanom kontajneri s nette/di 3.2.7. Nette Database vlastné callbacky po commite nemá; overte, že ich vaša služba transakcií volá len pre najvonkajšiu transakciu.

Náhodná zložka opakovania a služba `Random\Randomizer`

Od Hermesu 6.0 Dispatcher prijíma voliteľný ?Random\Randomizer $randomizer (pozrite Integrácia do frameworkov). Nette ho autowiruje: ak má kontajner službu Random\Randomizer, dispatcher ju dostane. Ak chcete zachovať predvolený bezpečný engine, na Hermese 6.0 alebo novšom zapíšte dispatcher takto:

yaml

	hermes.dispatcher:
		create: Tomaj\Hermes\Dispatcher(randomizer: null)

Na Hermese 5.x Nette zlyhá s Unable to pass specified arguments to Tomaj\Hermes\Dispatcher::__construct(), preto to tam vynechajte.

Celý zdrojový kód

Všetky súbory, s install.sh, e2e.sh a programom pre Supervisor, sú v examples/frameworks/nette na GitHube. Overené s nette/application 3.3.0, nette/di 3.2.6, contributte/console 0.11.0 a Tracy 2.12 na PHP 8.4, so všetkými drivermi.

Časté otázky

Prečo contributte/console a nie obyčajný PHP skript?

Príkazy dostávajú závislosti z kontajnera ako presentery a --priority aj nápovedu máte zadarmo. Funguje aj obyčajný bin/worker.php, ktorý vytvorí kontajner a zavolá handle(); pozrite návod na uvítací e-mail.

Môžem namiesto Tracy použiť Monolog?

Áno. Zaregistrujte službu Monolog\Logger a odstráňte riadok s TracyToPsrLoggerAdapter, aby sa autowiroval len jeden LoggerInterface. Pozrite Logovanie.

Kam dať driver pre jednotlivé prostredia?

Do config/local.neon, ktorý Bootstrap načíta ako posledný. Napríklad parameters: hermes: driver: rabbitmq. Ako si vybrať, nájdete v Prehľade driverov.

Čaká presenter na handler?

Nie. emit() len zapíše správu do fronty a vráti sa; worker ju spracuje neskôr. Pozrite Pojmy.

Ako spustiť viac workerov?

Zvýšte numprocs v programe pre Supervisor. Každý worker berie z tej istej fronty po jednej správe. Pozrite Nasadenie do produkcie.

Nette ​

1. Inštalácia ​

2. Konfigurácia služieb ​

3. Logovanie cez Tracy ​

4. Vloženie správy z presentera ​

5. Napíšte handler ​

6. Spustite workera ​

7. Reštart workerov pri nasadení ​

8. Otestujte producenta ​

Emitovanie po commite do databázy ​

Náhodná zložka opakovania a služba Random\Randomizer ​

Celý zdrojový kód ​

Časté otázky ​

Prečo contributte/console a nie obyčajný PHP skript? ​

Môžem namiesto Tracy použiť Monolog? ​

Kam dať driver pre jednotlivé prostredia? ​

Čaká presenter na handler? ​

Ako spustiť viac workerov? ​

Nette

1. Inštalácia

2. Konfigurácia služieb

3. Logovanie cez Tracy

4. Vloženie správy z presentera

5. Napíšte handler

6. Spustite workera

7. Reštart workerov pri nasadení

8. Otestujte producenta

Emitovanie po commite do databázy

Náhodná zložka opakovania a služba `Random\Randomizer`

Celý zdrojový kód

Časté otázky

Prečo contributte/console a nie obyčajný PHP skript?

Môžem namiesto Tracy použiť Monolog?

Kam dať driver pre jednotlivé prostredia?

Čaká presenter na handler?

Ako spustiť viac workerov?