Как записать сообщение в очередь rabbit из обработчика?

Question

[Светлана]

Всем привет! У меня вопрос по поводу сообщений в RabbitMQ!
В общем я из контроллера записываю собщение в очередь, далее обработчик ее обрабатывает и под конец должен запушить другое сообщение в другую очередь, но почему-то в UI rabbit второго сообщения я не вижу и соответственно второй обработчик не запускается.
Мои действия:
1) Установила symfony/amqp-messenger
2) С помощью гугла настроила 2 очереди (возможно проблема в этом)

transports:
            # https://symfony.com/doc/current/messenger.html#transport-configuration
            one_job:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    exchange:
                        name: one_job
                    queues:
                        one_job:
                            binding_keys:
                                - one_job
            two_job:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    exchange:
                        name: two_job
                    queues:
                        two_job:
                            binding_keys:
                                - two_job
        routing:
            'App\Message\SendOneMessage': one_job
            'App\Message\SendTwoMessage': two_job

3) Запустила php bin/console messenger:consume one_job и php bin/console messenger:consume two_job в разных терминалах
4) В контроллере отправила сообщение в очередь №1
$this->bus->dispatch(new SendOneMessage($id));
5) Проверила сообщение в UI RabbitMQ (Так же там указаны 2 подключения)
6) Далее отработал первый обработчки (Там у меня стояла точка остановы), он так же вызвал
$this->bus->dispatch(new SendTwoMessage($uuid));
7) Дальше я ждала что второе сообщение запишется в свою очередь и так же обработается, но ничего не происходит.

Подскажите пожалуйста, я правильно всё сделала? Если нет, то подскажите как правильно.

Comments

[tukreb]

А что в профайлере пишет? Там можно смотреть вызов событий и другие интересности.

[Светлана]

tukreb, что-то Message я там не нашла

[tukreb]

Светлана, странно, т.к он должен быть там.
Пробуйте консольный вариант
https://symfony.com/doc/current/messenger.html
https://symfonycasts.com/screencast/messenger/anot...
Ну или найдите Messages вкладку :)

[Светлана]

tukreb, спасибо! Нашла, но там только 1 сообщение SendOneMessage, там не видно что отправляется второе сообщение во вторую очередь.

[BoShurik]

Светлана, а если в MESSENGER_TRANSPORT_DSN передать sync:// второе сообщение будет выброшено?

[Светлана]

BoShurik, получю ошибку "You cannot receive messages from the Messenger SyncTransport."

###> symfony/messenger ###
MESSENGER_TRANSPORT_DSN=sync://user:password@amqp:5672/%2f/messages

[BoShurik]

Светлана,
MESSENGER_TRANSPORT_DSN=sync://
В этом случае запускать команду php bin/console messenger:consume не нужно, т.к. все будет выполнятся синхронно. Собственно, спрашиваю это чтоб проверить может проблема в коде

[Светлана]

BoShurik, без запуска команды php bin/console messenger:consume, сообщение отправляется, но обработчик не запускается. А как запуститится обработчик в таком случае? (без запуска консьюмера)

[BoShurik]

Светлана, В синхронном транспорте sync:// rabbitmq не используется, все обработчики запускаются синхронно. Если даже в этом случае не идет отправка второго сообщения, значит ошибка где-то в коде и надо его показать

[Светлана]

BoShurik, не судите строго) Никакой сложной логики, я пока только тестирую новую для себя технологию)

// Контроллер
public function __construct(private readonly MessageBusInterface $bus)
{
}

#[Route('/test-rabbit/{id}', name: 'create_image', methods: ['GET'])]
public function testRabbit(Image $image): array
{
    try {
        $this->bus->dispatch(new SendOneMessage($image->getId()));
    } catch (Exception $e) {
        return ['status' => false, 'message' => $e->getMessage()];
    }

    return ['status' => true];
}

// Обработчики
#[AsMessageHandler]
class OneJobHandler implements MessageHandlerInterface
{
    public function __construct(
        private readonly ImageRepository $imageRepository,
        private readonly MessageBusInterface $bus
    ) {
    }

    public function __invoke(SendOneMessage $message)
    {
        $image = $this->imageRepository->find($message->imageId);

        $image
            ->setDescription('Test')
            ->setDateUpdate(new DateTime())
        ;

        $this->imageRepository->save($image, true);

        $this->bus->dispatch(new SendTwoMessage($image->getUuid()));
    }
}

#[AsMessageHandler]
class TwoJobHandler implements MessageHandlerInterface
{
    public function __construct(
        private readonly TemplateRepository $templateRepository,
        private readonly SendTemplateMailer $sendTemplateMailer
    ) {
    }

    public function __invoke(SendTwoMessage $message)
    {
        $template = $this->templateRepository->findOneBy(['uuid' => $message->templateUuid]);

        $this->sendTemplateMailer->sendTemplate($template);
    }
}

// Messages

class SendOneMessage
{
    public int $imageId;

    public function __construct(int $imageId)
    {
        $this->imageId = $imageId;
    }
}

class SendTwoMessage
{
    public int $templateUuid;

    public function __construct(int $templateUuid)
    {
        $this->templateUuid = $templateUuid;
    }
}

Логика такая: Контроллер передает идентификатор изображения и передает его в rabbit, далее первый обработчки вносит изменения в сущность (что бы я могла понять, что обработчик выполнил работу). Далее кладу второе сообщение во вторую очередь с идентификатором шаблона. Второй обработчки достает этот шаблон по uuid и отправляет сообщение на email. В методе sendTemplate тоже элементраная логка, а имеено заполняется твиг шаблон и отправляется сообщение на почту с помощью symfony/mailer (этот метод рабочий).

[BoShurik]

Светлана, вроде бы все правильно. Единственное.
В конфиге у вас
App\Message\SendOneMessage
А в профайлере
App\Component\Message\SendOneMessage

[Светлана]

BoShurik, это переместилось во время разработки) В конфигах тоже актуальная информация.

[BoShurik]

Светлана, попробуйте для теста убрать всю полезную нагрузку и оставить только отправку сообщений. Может быть ошибка в каком-то сервисе и обработка останавливается

[Светлана]

BoShurik, провела тесты, получилось отправить второе сообщение. Сделала тест с изменением данных в БД, оба обработчика изменили свои таблицы. Пыталась отловить запуск сообщений спомощью LoggerInterface, но почему-то в обработчиках логи не пишутся :( и так же не отрабатывает метод sendTemplate, не подскажите почему так? Хотя если писать логи из контроллера, то всё нормально, так же с методом sendTemplate, он корректно работает из контроллера.

[BoShurik]

Светлана, без кода сложно какие-то выводы делать

[Светлана]

BoShurik, посмотрите пожалуйста:

// Контроллер
public function __construct(
    private readonly MessageBusInterface $bus,
    private LoggerInterface $logger,
    private readonly SendTemplateMailer $sendTemplateMailer,
    private readonly TemplateRepository $templateRepository
) {
}

#[Route('/test-rabbit/{id}', name: 'create_image', methods: ['GET'])]
public function testRabbit(Image $image): array
{
    $this->logger->info('Test 1'); // тут сообщение логируется

    $template = $this->templateRepository->findOneBy(['uuid' => $image->getUuid()]);
    $this->sendTemplateMailer->sendTemplate($template); // тут сообщение приходит на почту

    try {
        $this->bus->dispatch(new SendOneMessage($image->getId()));
    } catch (Exception $e) {
        return ['status' => false, 'message' => $e->getMessage()];
    }

    return ['status' => true];
}

// Обработчики

#[AsMessageHandler]
class OneJobHandler implements MessageHandlerInterface
{
    public function __construct(
        private readonly ImageRepository $imageRepository,
        private LoggerInterface $logger,
        private readonly MessageBusInterface $bus
    ) {
    }

    public function __invoke(SendOneMessage $message)
    {
        $this->logger->info('Test 2'); // в обработчике уже не логируется тут и дальше

        $image = $this->imageRepository->find($message->imageId);

        $image
            ->setDescription('Test')
            ->setDateUpdate(new DateTime())
        ;

        $this->imageRepository->save($image, true);

        $this->bus->dispatch(new SendTwoMessage($image->getUuid()));

        $this->logger->info('Test 3');
    }
}

#[AsMessageHandler]
class TwoJobHandler implements MessageHandlerInterface
{
    public function __construct(
        private readonly TemplateRepository $templateRepository,
        private LoggerInterface $logger,
        private readonly SendTemplateMailer $sendTemplateMailer
    ) {
    }

    public function __invoke(SendTwoMessage $message)
    {
        $this->logger->info('Test 4');

        $template = $this->templateRepository->findOneBy(['uuid' => $message->templateUuid]);

        $this->sendTemplateMailer->sendTemplate($template); // тут уже сообщение не отправляется, хотя тот же шаблон

        $this->logger->info('Test 5');
    }
}

// Mailer
class SendTemplateMailer
{
    public function __construct(
        private readonly Environment $twig,
        private readonly MailerInterface $mailer,
        private readonly string $emailFrom,
        private readonly LoggerInterface $logger
    ) {
    }
    
    public function sendTemplate(Template $template): void
    {
        $this->logger->info('Test 6'); // это сообщение появляется в логе, только когда метод вызывается из контроллера

        $email = (new Email())
            ->from($this->emailFrom)
            ->to($this->emailFrom)
            ->subject('Тестовое сообщение')
            ->html(
                $this
                    ->twig
                    ->load('/Email/test.html.twig')
                    ->render(
                        [
                            'uuid' => $template->getUuid(),
                            'title' => $template->getTitle(),
                            'description' => $template->getDescription()
                        ]
                    )
            )
        ;

        $this->mailer->send($email);
    }
}

[BoShurik]

Светлана,

// SendTwoMessage
public function __construct(int $templateUuid)

$template = $this->templateRepository->findOneBy(['uuid' => $message->templateUuid]);

Не уверен, что uuid имеет тип int, т.о. $template оказывается null и не может передасться в sendTemplate(Template $template)

[Светлана]

BoShurik, да, действительно ошиблась :) он имеет тип стринг (во всем виноват копипаст :) ). Исправила ошибку, но всё без изменений. Жаль, что там не срабатывает дебагер :(

[BoShurik]

Светлана, попробуйте примените к коду psalm, он вам на аналогичные ошибки укажет

[Светлана]

BoShurik, у меня стоит и psalm и phpstan и phpmd и phpcode_sniffer, но их надо запускать из консоли. Всё время забываю, а IDE настроить не удалось :(