Merge pull request #54 from Prokyonn/enhancement/async-producer

Prokyonn · web-flow · commit 3200770c7198 · 2025-03-06T10:10:14.000+01:00
Add TriggerProducer messages to make them optionally async
diff --git a/src/Command/SynchronizeProductVariantsCommand.php b/src/Command/SynchronizeProductVariantsCommand.php
@@ -14,20 +14,21 @@
 namespace Sulu\SyliusProducerPlugin\Command;
 
 use Doctrine\ORM\EntityManagerInterface;
-use Sulu\SyliusProducerPlugin\Producer\ProductVariantMessageProducerInterface;
+use Sulu\SyliusProducerPlugin\Message\TriggerProductVariantProducerMessage;
 use Sylius\Component\Core\Model\ProductVariantInterface;
 use Sylius\Component\Product\Repository\ProductVariantRepositoryInterface;
 use Symfony\Component\Console\Helper\ProgressBar;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Messenger\MessageBusInterface;
 
 class SynchronizeProductVariantsCommand extends BaseSynchronizeCommand
 {
     /** @param ProductVariantRepositoryInterface<ProductVariantInterface> $productVariantRepository */
     public function __construct(
         private EntityManagerInterface $entityManager,
-        private ProductVariantMessageProducerInterface $productVariantMessageProducer,
         private ProductVariantRepositoryInterface $productVariantRepository,
+        private MessageBusInterface $messageBus,
     ) {
         parent::__construct($entityManager);
     }
@@ -73,7 +74,7 @@ private function syncProductVariants(OutputInterface $output): void
                 continue;
             }
 
-            $this->productVariantMessageProducer->synchronize($productVariant);
+            $this->messageBus->dispatch(new TriggerProductVariantProducerMessage($productVariant->getId()));
 
             $this->entityManager->detach($productVariant);
             ++$processedItems;
diff --git a/src/Command/SynchronizeProductsCommand.php b/src/Command/SynchronizeProductsCommand.php
@@ -14,20 +14,21 @@
 namespace Sulu\SyliusProducerPlugin\Command;
 
 use Doctrine\ORM\EntityManagerInterface;
-use Sulu\SyliusProducerPlugin\Producer\ProductMessageProducerInterface;
+use Sulu\SyliusProducerPlugin\Message\TriggerProductProducerMessage;
 use Sylius\Component\Core\Model\ProductInterface;
 use Sylius\Component\Core\Repository\ProductRepositoryInterface;
 use Symfony\Component\Console\Helper\ProgressBar;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Messenger\MessageBusInterface;
 
 class SynchronizeProductsCommand extends BaseSynchronizeCommand
 {
     /** @param ProductRepositoryInterface<ProductInterface> $productRepository */
     public function __construct(
         private EntityManagerInterface $entityManager,
-        private ProductMessageProducerInterface $productMessageProducer,
         private ProductRepositoryInterface $productRepository,
+        private MessageBusInterface $messageBus,
     ) {
         parent::__construct($entityManager);
     }
@@ -72,8 +73,7 @@ private function syncProducts(OutputInterface $output): void
             if (!$product instanceof ProductInterface) {
                 continue;
             }
-
-            $this->productMessageProducer->synchronize($product);
+            $this->messageBus->dispatch(new TriggerProductProducerMessage($product->getId()));
 
             $this->entityManager->detach($product);
             ++$processedItems;
diff --git a/src/Command/SynchronizeTaxonCommand.php b/src/Command/SynchronizeTaxonCommand.php
@@ -14,19 +14,20 @@
 namespace Sulu\SyliusProducerPlugin\Command;
 
 use Doctrine\ORM\EntityManagerInterface;
-use Sulu\SyliusProducerPlugin\Producer\TaxonMessageProducerInterface;
+use Sulu\SyliusProducerPlugin\Message\TriggerTaxonProducerMessage;
 use Sylius\Component\Taxonomy\Model\TaxonInterface;
 use Sylius\Component\Taxonomy\Repository\TaxonRepositoryInterface;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Messenger\MessageBusInterface;
 
 class SynchronizeTaxonCommand extends BaseSynchronizeCommand
 {
     /** @param TaxonRepositoryInterface<TaxonInterface> $taxonRepository */
     public function __construct(
         private EntityManagerInterface $entityManager,
-        private TaxonMessageProducerInterface $taxonMessageProducer,
         private TaxonRepositoryInterface $taxonRepository,
+        private MessageBusInterface $messageBus,
     ) {
         parent::__construct($entityManager);
     }
@@ -62,7 +63,11 @@ private function syncTaxonTree(OutputInterface $output): void
             $taxons = \array_merge($taxons, $this->extractChildrenFlat($rootTaxon));
         }
 
-        $this->taxonMessageProducer->synchronize($taxons);
+        $taxonIds = \array_map(
+            fn (TaxonInterface $taxon) => $taxon->getId(),
+            $taxons,
+        );
+        $this->messageBus->dispatch(new TriggerTaxonProducerMessage($taxonIds));
     }
 
     private function extractChildrenFlat(TaxonInterface $rootTaxon): array
diff --git a/src/Message/TriggerProductProducerMessage.php b/src/Message/TriggerProductProducerMessage.php
@@ -0,0 +1,20 @@
+<?php
+
+/*
+ * This file is part of Sulu.
+ *
+ * (c) Sulu GmbH
+ *
+ * This source file is subject to the MIT license that is bundled
+ * with this source code in the file LICENSE.
+ */
+
+namespace Sulu\SyliusProducerPlugin\Message;
+
+class TriggerProductProducerMessage
+{
+    public function __construct(
+        public int $productId,
+    ) {
+    }
+}
diff --git a/src/Message/TriggerProductVariantProducerMessage.php b/src/Message/TriggerProductVariantProducerMessage.php
@@ -0,0 +1,20 @@
+<?php
+
+/*
+ * This file is part of Sulu.
+ *
+ * (c) Sulu GmbH
+ *
+ * This source file is subject to the MIT license that is bundled
+ * with this source code in the file LICENSE.
+ */
+
+namespace Sulu\SyliusProducerPlugin\Message;
+
+class TriggerProductVariantProducerMessage
+{
+    public function __construct(
+        public int $productVariantId,
+    ) {
+    }
+}
diff --git a/src/Message/TriggerTaxonProducerMessage.php b/src/Message/TriggerTaxonProducerMessage.php
@@ -0,0 +1,23 @@
+<?php
+
+/*
+ * This file is part of Sulu.
+ *
+ * (c) Sulu GmbH
+ *
+ * This source file is subject to the MIT license that is bundled
+ * with this source code in the file LICENSE.
+ */
+
+namespace Sulu\SyliusProducerPlugin\Message;
+
+class TriggerTaxonProducerMessage
+{
+    /**
+     * @param int[] $taxonIds
+     */
+    public function __construct(
+        public array $taxonIds,
+    ) {
+    }
+}
diff --git a/src/MessageHandler/TriggerProductProducerMessageHandler.php b/src/MessageHandler/TriggerProductProducerMessageHandler.php
@@ -0,0 +1,42 @@
+<?php
+
+/*
+ * This file is part of Sulu.
+ *
+ * (c) Sulu GmbH
+ *
+ * This source file is subject to the MIT license that is bundled
+ * with this source code in the file LICENSE.
+ */
+
+namespace Sulu\SyliusProducerPlugin\MessageHandler;
+
+use Sulu\SyliusProducerPlugin\Message\TriggerProductProducerMessage;
+use Sulu\SyliusProducerPlugin\Producer\ProductMessageProducerInterface;
+use Sylius\Component\Core\Model\ProductInterface;
+use Sylius\Component\Product\Repository\ProductRepositoryInterface;
+use Symfony\Component\Messenger\Attribute\AsMessageHandler;
+
+#[AsMessageHandler]
+class TriggerProductProducerMessageHandler
+{
+    /**
+     * @param ProductRepositoryInterface<ProductInterface> $productRepository
+     */
+    public function __construct(
+        private ProductRepositoryInterface $productRepository,
+        private ProductMessageProducerInterface $producer,
+    ) {
+    }
+
+    public function __invoke(TriggerProductProducerMessage $message): void
+    {
+        /** @var ProductInterface|null $product */
+        $product = $this->productRepository->find($message->productId);
+        if (null === $product) {
+            return;
+        }
+
+        $this->producer->synchronize($product);
+    }
+}
diff --git a/src/MessageHandler/TriggerProductVariantProducerMessageHandler.php b/src/MessageHandler/TriggerProductVariantProducerMessageHandler.php
@@ -0,0 +1,42 @@
+<?php
+
+/*
+ * This file is part of Sulu.
+ *
+ * (c) Sulu GmbH
+ *
+ * This source file is subject to the MIT license that is bundled
+ * with this source code in the file LICENSE.
+ */
+
+namespace Sulu\SyliusProducerPlugin\MessageHandler;
+
+use Sulu\SyliusProducerPlugin\Message\TriggerProductVariantProducerMessage;
+use Sulu\SyliusProducerPlugin\Producer\ProductVariantMessageProducerInterface;
+use Sylius\Component\Core\Model\ProductVariantInterface;
+use Sylius\Component\Core\Repository\ProductVariantRepositoryInterface;
+use Symfony\Component\Messenger\Attribute\AsMessageHandler;
+
+#[AsMessageHandler]
+class TriggerProductVariantProducerMessageHandler
+{
+    /**
+     * @param ProductVariantRepositoryInterface<ProductVariantInterface> $productVariantRepository
+     */
+    public function __construct(
+        private ProductVariantRepositoryInterface $productVariantRepository,
+        private ProductVariantMessageProducerInterface $producer,
+    ) {
+    }
+
+    public function __invoke(TriggerProductVariantProducerMessage $message): void
+    {
+        /** @var ProductVariantInterface|null $product */
+        $product = $this->productVariantRepository->find($message->productVariantId);
+        if (!$product) {
+            return;
+        }
+
+        $this->producer->synchronize($product);
+    }
+}
diff --git a/src/MessageHandler/TriggerTaxonProducerMessageHandler.php b/src/MessageHandler/TriggerTaxonProducerMessageHandler.php
@@ -0,0 +1,42 @@
+<?php
+
+/*
+ * This file is part of Sulu.
+ *
+ * (c) Sulu GmbH
+ *
+ * This source file is subject to the MIT license that is bundled
+ * with this source code in the file LICENSE.
+ */
+
+namespace Sulu\SyliusProducerPlugin\MessageHandler;
+
+use Sulu\SyliusProducerPlugin\Message\TriggerTaxonProducerMessage;
+use Sulu\SyliusProducerPlugin\Producer\TaxonMessageProducerInterface;
+use Sylius\Component\Core\Model\TaxonInterface;
+use Sylius\Component\Taxonomy\Repository\TaxonRepositoryInterface;
+use Symfony\Component\Messenger\Attribute\AsMessageHandler;
+
+#[AsMessageHandler]
+class TriggerTaxonProducerMessageHandler
+{
+    /**
+     * @param TaxonRepositoryInterface<TaxonInterface> $taxonRepository
+     */
+    public function __construct(
+        private TaxonRepositoryInterface $taxonRepository,
+        private TaxonMessageProducerInterface $producer,
+    ) {
+    }
+
+    public function __invoke(TriggerTaxonProducerMessage $message): void
+    {
+        /** @var TaxonInterface[] $taxon */
+        $taxon = $this->taxonRepository->findBy(['id' => $message->taxonIds]);
+        if (0 === \count($taxon)) {
+            return;
+        }
+
+        $this->producer->synchronize($taxon);
+    }
+}
diff --git a/src/Resources/config/services.xml b/src/Resources/config/services.xml
@@ -30,6 +30,36 @@
             <argument type="service" id="sulu_sylius_producer.messenger_bus"/>
         </service>
 
+        <service
+            id="Sulu\SyliusProducerPlugin\MessageHandler\TriggerProductProducerMessageHandler"
+            class="Sulu\SyliusProducerPlugin\MessageHandler\TriggerProductProducerMessageHandler"
+        >
+            <argument type="service" id="sylius.repository.product"/>
+            <argument type="service" id="Sulu\SyliusProducerPlugin\Producer\ProductMessageProducerInterface"/>
+
+            <tag name="messenger.message_handler"/>
+        </service>
+
+        <service
+            id="Sulu\SyliusProducerPlugin\MessageHandler\TriggerProductVariantProducerMessageHandler"
+            class="Sulu\SyliusProducerPlugin\MessageHandler\TriggerProductVariantProducerMessageHandler"
+        >
+            <argument type="service" id="sylius.repository.product_variant"/>
+            <argument type="service" id="Sulu\SyliusProducerPlugin\Producer\ProductVariantMessageProducerInterface"/>
+
+            <tag name="messenger.message_handler"/>
+        </service>
+
+        <service
+            id="Sulu\SyliusProducerPlugin\MessageHandler\TriggerTaxonProducerMessageHandler"
+            class="Sulu\SyliusProducerPlugin\MessageHandler\TriggerTaxonProducerMessageHandler"
+        >
+            <argument type="service" id="sylius.repository.taxon"/>
+            <argument type="service" id="Sulu\SyliusProducerPlugin\Producer\TaxonMessageProducerInterface"/>
+
+            <tag name="messenger.message_handler"/>
+        </service>
+
         <service id="Sulu\SyliusProducerPlugin\EventSubscriber\ProductEventSubscriber">
             <argument type="service" id="Sulu\SyliusProducerPlugin\Producer\ProductMessageProducerInterface"/>
 
@@ -72,24 +102,24 @@
 
         <service id="Sulu\SyliusProducerPlugin\Command\SynchronizeTaxonCommand">
             <argument type="service" id="doctrine.orm.entity_manager"/>
-            <argument type="service" id="Sulu\SyliusProducerPlugin\Producer\TaxonMessageProducerInterface"/>
             <argument type="service" id="sylius.repository.taxon"/>
+            <argument type="service" id="sulu_sylius_producer.messenger_bus"/>
 
             <tag name="console.command" />
         </service>
 
         <service id="Sulu\SyliusProducerPlugin\Command\SynchronizeProductsCommand">
             <argument type="service" id="doctrine.orm.entity_manager"/>
-            <argument type="service" id="Sulu\SyliusProducerPlugin\Producer\ProductMessageProducerInterface"/>
             <argument type="service" id="sylius.repository.product"/>
+            <argument type="service" id="sulu_sylius_producer.messenger_bus"/>
 
             <tag name="console.command" />
         </service>
 
         <service id="Sulu\SyliusProducerPlugin\Command\SynchronizeProductVariantsCommand">
             <argument type="service" id="doctrine.orm.entity_manager"/>
-            <argument type="service" id="Sulu\SyliusProducerPlugin\Producer\ProductVariantMessageProducerInterface"/>
             <argument type="service" id="sylius.repository.product_variant"/>
+            <argument type="service" id="sulu_sylius_producer.messenger_bus"/>
 
             <tag name="console.command" />
         </service>
