magento/adobe-stock-integration#1724: Support batches processing for synchronization queue messages - implemented media content sychronization

Author: yolouiese

app/code/Magento/MediaContentSynchronization/Model/Consume.php

@@ -7,8 +7,9 @@
 
 namespace Magento\MediaContentSynchronization\Model;
 
+use Magento\AsynchronousOperations\Api\Data\OperationInterface;
 use Magento\Framework\Exception\LocalizedException;
-use Magento\MediaContentApi\Api\Data\ContentIdentityInterface;
+use Magento\Framework\Serialize\SerializerInterface;
 use Magento\MediaContentSynchronizationApi\Api\SynchronizeIdentitiesInterface;
 use Magento\MediaContentSynchronizationApi\Api\SynchronizeInterface;
 
@@ -17,6 +18,11 @@
  */
 class Consume
 {
+    /**
+     * @var SerializerInterface
+     */
+    private $serializer;
+
     /**
      * @var SynchronizeInterface
      */
@@ -28,24 +34,30 @@ class Consume
     private $synchronizeIdentities;
 
     /**
+     * @param SerializerInterface $serializer
      * @param SynchronizeInterface $synchronize
      * @param SynchronizeIdentitiesInterface $synchronizeIdentities
      */
     public function __construct(
+        SerializerInterface $serializer,
         SynchronizeInterface $synchronize,
         SynchronizeIdentitiesInterface $synchronizeIdentities
     ) {
+        $this->serializer = $serializer;
         $this->synchronize = $synchronize;
         $this->synchronizeIdentities = $synchronizeIdentities;
     }
 
     /**
      * Run media files synchronization.
-     * @param string[] $identities
+     * @param OperationInterface $operation
      * @throws LocalizedException
      */
-    public function execute(array $identities) : void
+    public function execute(OperationInterface $operation) : void
     {
+        $serializedData = $operation->getSerializedData();
+        $identities = $this->serializer->unserialize($serializedData);
+
         if (!empty($identities)) {
             $this->synchronizeIdentities->execute($identities);
         } else {

app/code/Magento/MediaContentSynchronization/Model/Publish.php

@@ -7,7 +7,11 @@
 
 namespace Magento\MediaContentSynchronization\Model;
 
+use Magento\AsynchronousOperations\Api\Data\OperationInterfaceFactory;
+use Magento\Framework\Bulk\OperationInterface;
+use Magento\Framework\DataObject\IdentityGeneratorInterface;
 use Magento\Framework\MessageQueue\PublisherInterface;
+use Magento\Framework\Serialize\SerializerInterface;
 
 /**
  * Publish media content synchronization queue.
@@ -19,16 +23,41 @@ class Publish
      */
     private const TOPIC_MEDIA_CONTENT_SYNCHRONIZATION = 'media.content.synchronization';
 
+    /**
+     * @var OperationInterfaceFactory
+     */
+    private $operationFactory;
+
+    /**
+     * @var IdentityGeneratorInterface
+     */
+    private $identityService;
+
     /**
      * @var PublisherInterface
      */
     private $publisher;
 
     /**
+     * @var SerializerInterface
+     */
+    private $serializer;
+
+    /**
+     * @param OperationInterfaceFactory $operationFactory
+     * @param IdentityGeneratorInterface $identityService
      * @param PublisherInterface $publisher
+     * @param SerializerInterface $serializer
      */
-    public function __construct(PublisherInterface $publisher)
-    {
+    public function __construct(
+        OperationInterfaceFactory $operationFactory,
+        IdentityGeneratorInterface $identityService,
+        PublisherInterface $publisher,
+        SerializerInterface $serializer
+    ) {
+        $this->operationFactory = $operationFactory;
+        $this->identityService = $identityService;
+        $this->serializer = $serializer;
         $this->publisher = $publisher;
     }
 
@@ -38,9 +67,22 @@ public function __construct(PublisherInterface $publisher)
      */
     public function execute(array $contentIdentities = []) : void
     {
+        $bulkUuid = $this->identityService->generateId();
+        $dataToEncode = $this->serializer->serialize($contentIdentities);
+
+        $data = [
+            'data' => [
+                'bulk_uuid' => $bulkUuid,
+                'topic_name' => self::TOPIC_MEDIA_CONTENT_SYNCHRONIZATION,
+                'serialized_data' => $dataToEncode,
+                'status' => OperationInterface::STATUS_TYPE_OPEN,
+            ]
+        ];
+        $operation = $this->operationFactory->create($data);
+
         $this->publisher->publish(
             self::TOPIC_MEDIA_CONTENT_SYNCHRONIZATION,
-            $contentIdentities
+            $operation
         );
     }
 }

app/code/Magento/MediaContentSynchronization/etc/communication.xml

@@ -7,7 +7,7 @@
 -->
 <config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:noNamespaceSchemaLocation="urn:magento:framework:Communication/etc/communication.xsd">
-    <topic name="media.content.synchronization" is_synchronous="false" request="string[]">
+    <topic name="media.content.synchronization" is_synchronous="false" request="Magento\AsynchronousOperations\Api\Data\OperationInterface">
         <handler name="media.content.synchronization.handler"
                  type="Magento\MediaContentSynchronization\Model\Consume" method="execute"/>
     </topic>