Как настроить symfony\messenger вне symfony?

Question

[dinya17]

У меня есть задача - вынести огромный кусок логики и кода в пакет для переиспользования.
Но для тестирования и работы этого кода требуется настройка symfony\messenger.
Изначально вся логика кода была написана и опробована внутри symfony и работало все хорошо.
Но вот при попытке настроить подобную же среду в пакете для тестирования кода я столкнулся с проблемой.
Очень сложно настроить пакет symfony\messenger вне symfony.
есть небольшие инструкции про синхронную очередь и на этом все.
А мне требуется 1 синхронная и 2 асинхронные.
На хабре даже статью находил, но там все примитивно, чего не достаточно для меня.

Задача настройки взаимодействия 3 очередей вне симфони оказалась очень сложным делом.
Может у кого то есть опыт или ссылка нормальную инструкцию как это сделать?

Так же столкнулся с проблемой, что требуется DI для этого.
И вот он выдается что есть циклические зависимости, когда 1 хендлер обрабатывает сообщение из очереди и отправляет ее в другую синхронную или асинхронную.

Comments

[BoShurik]

Тут были: https://symfony.com/doc/current/components/messeng... ?

[dinya17]

С этого и начал.
Вся штука в том, что там описан простейший способ настройки синхронной шины.
А вот если начать комбинировать шины, то начинается засада.

[BoShurik]

dinya17, сейчас ваш вопрос выглядит максимально абстрактно. Попробуйте конкретизировать. В идеале проект на гитхаб выложите, тогда можно будет что-то конкретное посоветовать

[Виктор]

dinya17, засада с какими войсками?
Что значит "синхронная/асинхронная шина"? Симфони шину не предоставляет, он предоставляет инструмент посыла сообщений и их прослушивания.

[dinya17]

хмм, 3 шины.
1 работает через синхронный транспорт
2 работает через асинхронный транспорт
3 работает через асинхронный транспорт

Задача такая.
Отправлять сообщения в одну шину AsyncBusA , работающую через асинхронный транспорт.
Это сообщение получается в AsyncBusA в этой шине через шину в синхронном режиме выполняем сообщение какое - то.
Теперь надо отправить изAsyncBusA сообщение в AsyncBusB

В AsyncBusВ мы получаем сообщение что - то вычисляем для себя и снова отправляем сообщение в AsyncBusА.

[dinya17]

Прочитал свой комментарий. В общем картинка правильно все показывает. Текст запутан.

[dinya17]

Многочисленные попытки пришли примерно к этому, но все равно есть проблемы в виде циклических зависимостей.

$builder = new ContainerBuilder();
        $definitions = [
            ServiceIds::TRANSPORT_SYNC => static function (ContainerInterface $container) {
                return new SyncTransport($container->get(ServiceIds::SYNC_BUS));
            },
            ServiceIds::TRANSPORT_ACTIVITY => new InMemoryTransport(),
            ServiceIds::TRANSPORT_WORKFLOW => new InMemoryTransport(),
            EventDispatcherInterface::class => new EventDispatcher(),
            WorkflowRepositoryInterface::class => new WorkflowRepository(),
            ActivityRepositoryInterface::class => new ActivityRepository(),
            ServiceIds::SYNC_BUS => factory(
                static function (ContainerInterface $container): MessageBus {
                    return new MessageBus(
                        [
                            new HandleMessageMiddleware(
                                new HandlersLocator(
                                    [
                                        NotifyToMattermostMessage1::class => [new NotifyToMattermostMessageHandler1()],
                                        ActivityMessage::class => [
                                            $container->get(ActivityMessageHandler::class)
                                        ],
                                        WorkflowMessage::class => [
                                            new WorkflowMessageHandler(
                                                $container->get(WorkflowRepositoryInterface::class),
                                                $container->get(EventDispatcherInterface::class),
                                                $container->get(ActivityRepositoryInterface::class),
                                                $container->get(ServiceIds::ACTIVITY_BUS),
                                            ),
                                        ]
                                    ]
                                )
                            ),
                        ]
                    );
                }
            ),
            ServiceIds::WORKFLOW_BUS => factory(
                static function (ContainerInterface $container): MessageBus {
                    return new MessageBus(
                        [
                            new SendMessageMiddleware(
                                new SendersLocator(
                                    [
                                        WorkflowMessage::class => [ServiceIds::TRANSPORT_WORKFLOW]
                                    ],
                                    $container
                                ),
                                $container->get(EventDispatcherInterface::class)
                            ),
                            new HandleMessageMiddleware(
                                new HandlersLocator(
                                    [
                                        WorkflowMessage::class => [
                                            new WorkflowMessageHandler(
                                                $container->get(WorkflowRepositoryInterface::class),
                                                $container->get(EventDispatcherInterface::class),
                                                $container->get(ActivityRepositoryInterface::class),
                                                $container->get(ServiceIds::ACTIVITY_BUS),
                                            ),
                                        ],
                                    ]
                                )
                            ),
                        ]
                    );
                }
            ),
            ServiceIds::ACTIVITY_BUS => factory(
                static function (ContainerInterface $container): MessageBus {

                    return new MessageBus(
                        [
                            new SendMessageMiddleware(
                                new SendersLocator(
                                    [
                                        ActivityMessage::class => [ServiceIds::TRANSPORT_ACTIVITY]
                                    ],
                                    $container
                                ),
                                $container->get(EventDispatcherInterface::class)
                            ),
                            new HandleMessageMiddleware(
                                new HandlersLocator(
                                    [
                                        ActivityMessage::class => [
                                            $container->get(ActivityMessageHandler::class)
                                        ],
                                    ]
                                )
                            ),
                        ]
                    );
                }
            ),

            ActivityMessageHandler::class => \DI\create(ActivityMessageHandler::class)
                 ->constructor(
                     get(ActivityRepositoryInterface::class),
                     get(EventDispatcherInterface::class),
                     get(ServiceIds::WORKFLOW_BUS),
                     get(ServiceIds::SYNC_BUS),
                 )
                 ->lazy(),

            WorkflowService::class => factory(
                static function (ContainerInterface $container): WorkflowService {
                    return new WorkflowService(
                        $container->get(WorkflowRepositoryInterface::class),
                        $container->get(ServiceIds::WORKFLOW_BUS)
                    );
                }
            ),

        ];

        $builder->addDefinitions($definitions);
        $container = $builder->build();

[dinya17]

Есть предположения?

[BoShurik]

dinya17,

Так же столкнулся с проблемой, что требуется DI для этого.

На каком моменте он требуется? Видится, будто бы можно и без него обойтись и вся проблема как раз в его настройке

[dinya17]

BoShurik, я выше привел конфиг. В нем мой код и там же написано ContainerInterface $container
Это и есть DI.
Вот где я его использую, там он требуется, сам Mesenger требует DI.
Без него не могу запустить, а с ним получается циклическая зависимость.

[BoShurik]

dinya17,

сам Mesenger требует DI

Откуда вы это взяли?

https://github.com/symfony/messenger/blob/7.0/comp... - тут требований нет
https://symfony.com/doc/current/components/messeng... - здесь тоже не нашел

[dinya17]

BoShurik, вот отсюда.

[BoShurik]

dinya17, это же не значит, что надо все конфигурировать через DI. Варианты навскидку:
- имплементировать свой SendersLocatorInterface без использования контейнера
- имплементировать свой ContainerInterface
- взять

Symfony\Component\DependencyInjection\ServiceLocator

в качестве локатора