Установка

1

$ composer require symfony/messenger

Концепції

Відправник:

Відповідає за серіалізацію та відправку повідомлень чомусь. Це щось може бути брокером повідомлнень або стороннім API, наприклад.

Отримувач:

Відповідає за десеріалізацію та перенаправлення повідомлень обробнику(ам). Це може бути пулер черги повідомлень або кінцева точка API, наприклад.

Обробник:

Відповідає за обробку повідомлень, використовуючи бізнес-логіку, що можна застосувати до повідомлень. Обробники викликаються проміжковим ПЗ HandleMessageMiddleware.

Проміжкове ПЗ:

Проміжкове ПЗ може отримати доступ до повідолення і його обгортки (конверту) під час запуску через автобус. В буквальному сенсі "програмне забезпечення посередині", воно не стосується основної функціональності (бізнес-логіки) додатку. Замість цього, воно скорочує шлях та може застосовуватися по всьому додатку, впливаючи на весь автобус повідомлень. Наприклад: ведення логів, валідацію повідомлення, початок транзакції, ... Воно також відповідальне за виклик наступного проміжкового ПЗ в ланцюжку, що означає, що воно також може налаштовувати конверт, додаючи до нього штампи або навіть переміщуючи його, а також перервати ланцюжок проміжкового ПЗ. Проміжкове ПЗ викликається при початковому оголошенні повідомлення, і пізніше, коли повідомлення отримано від транспорту.

Конверт:

Спеціальна концепнція месенджеру, що надає повну гнучкість всередині автобусу з повідомленнями, обгортаючи в себе повідомлення, та дозволяюючи додавати корисну інформацію всередині шляхом марок на конвертах.

Штампи на конвертах:

Інформація, яку вам потрібно додати до повідомлення: контекст серіалізатору, який використовувати для транспорту, маркери, що вказують на отримання повідомлення, або будь-які інші метадані, які може використовувати ваше проміжкове ПЗ або шар транспорту.

Автобус

Автобус використовується для оголошення повідомлень. Поведінка автобусу впорядкована в сполучному стеці. Компонент постачається з набором сполучного ПЗ, яке ви можете використовувати.

При використанні автобусу повідомлень з пакетом Symfony FrameworkBundle, для вас конфігурується наступне проміжкове ПЗ:

1. SendMessageMiddleware (підключає асинхронну обробку, веде логи обробки ваших повідомлень, якщо ви передасте логер)
2. HandleMessageMiddleware (викликає зареєстрованого(их) обробника(ів))

Приклад:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

use App\Message\MyMessage;
use App\MessageHandler\MyMessageHandler;
use Symfony\Component\Messenger\Handler\HandlersLocator;
use Symfony\Component\Messenger\MessageBus;
use Symfony\Component\Messenger\Middleware\HandleMessageMiddleware;

$handler = new MyMessageHandler();

$bus = new MessageBus([
    new HandleMessageMiddleware(new HandlersLocator([
        MyMessage::class => [$handler],
    ])),
]);

$bus->dispatch(new MyMessage(/* ... */));

Обробники

Після диспетчерізації в автобус, повідомлення будуть оброблені "обробниками повідомлень". Обробник повідомлень - це PHP-викликане (тобто функція або екземпляр класу), яке буде проводити необхідну обробку для вашого повідомлення:

1
2
3
4
5
6
7
8
9
10
11

namespace App\MessageHandler;

use App\Message\MyMessage;

class MyMessageHandler
{
    public function __invoke(MyMessage $message): void
   {
       // Обработка сообщения...
   }
}

Додавання метаданих до повідомлення (конверти)

Якщо вам необхідно додати метадані або деяку конфігурацію в повідомлення, обгорніть його в клас Envelope та додайте штампи. Наприклад, щоб встановити групи серіалізації, що використовуються при проходженні повідомлення через шар транспорту, використовуйте штамп SerializerStamp:

1
2
3
4
5
6
7
8
9
10

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Stamp\SerializerStamp;

$bus->dispatch(
    (new Envelope($message))->with(new SerializerStamp([
        // групи застосовується до всього повідомлення, тому не забудьте
        // визначити групу для кожного вбудованого об'єкту
        'groups' => ['my_serialization_groups'],
    ]))
);

Ось деякі важливі штампки на конверти, які відправляються в Symfony Messenger:

• DelayStamp,

  щоб відкласти обробку асинхронного повідомлення.

• DispatchAfterCurrentBusStamp,

  щоб повідомлення було оброблено після того, як буде виконаний поточний автобус. Прочитайте більше в Транзакційні повідомлення: обробляйте повідомлення після того, як обробку завершено.

• HandledStamp,

  штамп, який відмічає повідомлення обробленими конкретним обробником. Дозволяє отримати доступ до значення, що повернено обробником, та назві обробнику.

• ReceivedStamp,

  внутрішній штамп, який відмічає повідомлення отриманим від транспорту.

• SentStamp,

  штамп, який відмічає повідомлення відправленим від конкретного відправника. Дозволяє отримати доступ до FQCN відправника та псевдонім, якщо він доступний, з SendersLocator.

• SerializerStamp,

  щоб сконфігурувати групи серіалізації, що використовуються транспортом.

• ValidationStamp,

  щоб сконфігурувати групи валідації, що використовуються при увімкненнні проміжкового ПЗ валідації.

• ErrorDetailsStamp,

  внутрішній штамп, коли повідомлення зазнає невдачі у звʼязку з виключенням в обробнику.

• ScheduledStamp, штамп, який позначає повідомлення як створене планувальником. Це допомагає відрізнити його від повідомлень, створених "вручну". Ви можете дізнатися більше про це у Документації про планувальник.

Замість того, щоб розбиратися з повідомленнями напряму в проміжковому ПЗ, ви отримуєте конверт. Таким чином, ви можете досліджувати зміст конверту та його штампу, або додати їх:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

use App\Message\Stamp\AnotherStamp;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
use Symfony\Component\Messenger\Middleware\StackInterface;
use Symfony\Component\Messenger\Stamp\ReceivedStamp;

class MyOwnMiddleware implements MiddlewareInterface
{
    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        if (null !== $envelope->last(ReceivedStamp::class)) {
            // Повідомлення було щойно отримане...

            // Ви можете, наприклад, додати інший штамп.
            $envelope = $envelope->with(new AnotherStamp(/* ... */));
        } else {
            // Повідомлення було щойно оголошено
        }

        return $stack->next()->handle($envelope, $stack);
    }
}

Приклад вище перенаправить повідомлення до наступного проміжкового ПЗ з додатковим штампом якщо повідомлення було щойно отримане (тобто, має хоча б один штамп ReceivedStamp). Ви можете створювати власні штампи, реалізуючи StampInterface.

Якщо ви хочете дослідити всі штампи на конверті, використовуйте метод $envelope->all(), який повертає всі штампи, згруповані за типами (FQCN). Як варіант, ви можете ітерувати всі штампи певного типу, використовуючи FQCN в якості першого параметру цього методу (наприклад, $envelope->all(ReceivedStamp::class)).

Транспорт

Для відправки та отримання повідомлень вам знадобиться сконфігурувати транспорт. Транспорт відповідатиме за комунікацію з вашим брокером повідомлень або третіми сторонами.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

namespace App\MessageSender;

use App\Message\ImportantAction;
use Symfony\Component\Mailer\MailerInterface;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
use Symfony\Component\Mime\Email;

class ImportantActionToEmailSender implements SenderInterface
{
    public function __construct(
        private MailerInterface $mailer,
        private string $toEmail,
    ) {
    }

    public function send(Envelope $envelope): Envelope
    {
        $message = $envelope->getMessage();

        if (!$message instanceof ImportantAction) {
            throw new \InvalidArgumentException(sprintf('This transport only supports "%s" messages.', ImportantAction::class));
        }

        $this->mailer->send(
            (new Email())
                ->to($this->toEmail)
                ->subject('Important action made')
                ->html('<h1>Important action</h1><p>Made by '.$message->getUsername().'</p>')
        );

        return $envelope;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

namespace App\MessageReceiver;

use App\Message\NewOrder;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
use Symfony\Component\Serializer\SerializerInterface;

class NewOrdersFromCsvFileReceiver implements ReceiverInterface
{
    public function __construct(
        private SerializerInterface $serializer,
        private string $filePath,
    ) {
    }

    public function get(): iterable
    {
        // Отримайте конверт у відповідності до вашого транспорту, у більшості випадків,
        // ($yourEnvelope тут), використання зв'язку є найпростішим рішенням
        if (null === $yourEnvelope) {
            return [];
        }

        try {
            $envelope = $this->serializer->decode([
                'body' => $yourEnvelope['body'],
                'headers' => $yourEnvelope['headers'],
            ]);
        } catch (MessageDecodingFailedException $exception) {
            $this->connection->reject($yourEnvelope['id']);
            throw $exception;
        }

        return [$envelope->with(new CustomStamp($yourEnvelope['id']))];
    }

    public function ack(Envelope $envelope): void
    {
        // Додайте інформацію про оброблене повідомлення
    }

    public function reject(Envelope $envelope): void
    {
        // У випадку користувацього з'єднання
        $this->connection->reject($this->findCustomStamp($envelope)->getId());
    }
}

Дізнайтеся більше