Установка

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)
   {
       // Обработка сообщения...
   }
}

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

Якщо вам необхідно додати метадані або деяку конфігурацію в повідомлення, обгорніть його в клас 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:

1. DelayStamp, щоб відкласти обробку асинхронного повідомлення.
2. DispatchAfterCurrentBusStamp, щоб повідомлення було оброблено після того, як буде виконаний поточний автобус. Прочитайте більше в Транзакційні повідомлення: обробляйте повідомлення після того, як обробку завершено.
3. HandledStamp, штамп, який відмічає повідомлення обробленими конкретним обробником. Дозволяє отримати доступ до значення, що повернено обробником, та назві обробнику.
4. ReceivedStamp, внутрішній штамп, який відмічає повідомлення отриманим від транспорту.
5. SentStamp, штамп, який відмічає повідомлення відправленим від конкретного відправника. Дозволяє отримати доступ до FQCN відправника та псевдонім, якщо він доступний, з SendersLocator.
6. SerializerStamp, щоб сконфігурувати групи серіалізації, що використовуються транспортом.
7. ValidationStamp, щоб сконфігурувати групи валідації, що використовуються при увімкненнні проміжкового ПЗ валідації.
8. ErrorDetailsStamp, внутрішній штамп, коли повідомлення зазнає невдачі у звʼязку з виключенням в обробнику.

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

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
35
36
37

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
{
    private $mailer;
    private $toEmail;

    public function __construct(MailerInterface $mailer, string $toEmail)
    {
        $this->mailer = $mailer;
        $this->toEmail = $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
49
50
51

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
{
    private $serializer;
    private $filePath;

    public function __construct(SerializerInterface $serializer, string $filePath)
    {
        $this->serializer = $serializer;
        $this->filePath = $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());
    }
}

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