Як створити ваш власний транспорт повідомлень

Дата оновлення перекладу 2022-11-21

Як створити ваш власний транспорт повідомлень

Коли ви написали ваших відправника та отримувача транспорту, ви можете зареєструвати свою фабрику транспорту, щоб використовувати її через DSN у додатку Symfony.

Створіть вашу фабрику транспорту

Вам потрібно надати FrameworkBundle можливіть створювати ваш транспорт з DSN. Вам знадобиться фабрика транспорту:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
use Symfony\Component\Messenger\Transport\TransportFactoryInterface;
use Symfony\Component\Messenger\Transport\TransportInterface;

class YourTransportFactory implements TransportFactoryInterface
{
    public function createTransport(string $dsn, array $options, SerializerInterface $serializer): TransportInterface
    {
        return new YourTransport(/* ... */);
    }

    public function supports(string $dsn, array $options): bool
    {
        return 0 === strpos($dsn, 'my-transport://');
    }
}

Обʼект транспорту повинен реалізовувати TransportInterface (який поєднує SenderInterface і ReceiverInterface). Ось спрощений приклад транспорту БД:

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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;
use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
use Symfony\Component\Messenger\Transport\TransportInterface;
use Symfony\Component\Uid\Uuid;

class YourTransport implements TransportInterface
{
    private $db;
    private $serializer;

    /**
     * @param FakeDatabase $db is used for demo purposes. It is not a real class.
     */
    public function __construct(FakeDatabase $db, SerializerInterface $serializer = null)
    {
        $this->db = $db;
        $this->serializer = $serializer ?? new PhpSerializer();
    }

    public function get(): iterable
    {
        // Отримати повідомлення з "my_queue"
        $row = $this->db->createQuery(
                'SELECT *
                FROM my_queue
                WHERE (delivered_at IS NULL OR delivered_at < :redeliver_timeout)
                AND handled = FALSE'
            )
            ->setParameter('redeliver_timeout', new DateTimeImmutable('-5minutes'))
            ->getOneOrNullResult();

        if (null === $row) {
            return [];
        }

        $envelope = $this->serializer->decode([
            'body' => $row['envelope'],
        ]);

        return [$envelope->with(new TransportMessageIdStamp($row['id']))];
    }

    public function ack(Envelope $envelope): void
    {
        $stamp = $envelope->last(TransportMessageIdStamp::class);
        if (!$stamp instanceof TransportMessageIdStamp) {
            throw new \LogicException('No TransportMessageIdStamp found on the Envelope.');
        }

        // Відмітити повідомлення як "handled"
        $this->db->createQuery('UPDATE my_queue SET handled = TRUE WHERE id = :id')
            ->setParameter('id', $stamp->getId())
            ->execute();
    }

    public function reject(Envelope $envelope): void
    {
        $stamp = $envelope->last(TransportMessageIdStamp::class);
        if (!$stamp instanceof TransportMessageIdStamp) {
            throw new \LogicException('No TransportMessageIdStamp found on the Envelope.');
        }

        // Видалити повідомлення з таблиці "my_queue"
        $this->db->createQuery('DELETE FROM my_queue WHERE id = :id')
            ->setParameter('id', $stamp->getId())
            ->execute();
    }

    public function send(Envelope $envelope): Envelope
    {
        $encodedMessage = $this->serializer->encode($envelope);
        $uuid = (string) Uuid::v4();
        // Додати повідомлення до таблиці "my_queue"
        $this->db->createQuery(
                'INSERT INTO my_queue (id, envelope, delivered_at, handled)
                VALUES (:id, :envelope, NULL, FALSE)'
            )
            ->setParameters([
                'id' => $uuid,
                'envelope' => $encodedMessage['body'],
            ])
            ->execute();

        return $envelope->with(new TransportMessageIdStamp($uuid));
    }
}

Реалізація вище - невиконуваний код, але він ілюструє, як може бути реалізований TransportInterface. Шоб побачити справжні реалізації, див. InMemoryTransport і DoctrineReceiver.

Зареєструйте вашу фабрику

  • YAML
  • XML
  • PHP
1
2
3
4
# config/services.yaml
services:
    Your\Transport\YourTransportFactory:
        tags: [messenger.transport_factory]

Використайте ваш транспорт

У конфігурації framework.messenger.transports.*, створіть ваш названий транспорт, використовуючи власну DSN:

  • YAML
  • XML
  • PHP
1
2
3
4
5
# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            yours: 'my-transport://...'

Крім того, що ви зможете маршрутизувати ваші повідомлення відправнику yours, це надасть вам доступ до наступних сервісів:

  1. messenger.sender.yours: відправник;
  2. messenger.receiver.yours: отримувач.