magento-engcom/bulk-api#4 Support for Async operations in WebAPI

- changed interface names per architectural feedback
- added statuses tracking for successfull operations

Author: vrann

app/code/Magento/AsynchronousOperations/Api/BulkStatusInterface.php

@@ -20,7 +20,7 @@ interface BulkStatusInterface extends \Magento\Framework\Bulk\BulkStatusInterfac
      * Get Bulk summary data with list of operations items full data.
      *
      * @param string $bulkUuid
-     * @return \Magento\AsynchronousOperations\Api\Data\DetailedBulkStatusInterface
+     * @return \Magento\AsynchronousOperations\Api\Data\DetailedBulkOperationsStatusInterface
      * @throws \Magento\Framework\Exception\NoSuchEntityException
      */
     public function getBulkDetailedStatus($bulkUuid);
@@ -29,7 +29,7 @@ public function getBulkDetailedStatus($bulkUuid);
      * Get Bulk summary data with list of operations items short data.
      *
      * @param string $bulkUuid
-     * @return \Magento\AsynchronousOperations\Api\Data\BulkStatusInterface
+     * @return \Magento\AsynchronousOperations\Api\Data\BulkOperationsStatusInterface
      * @throws \Magento\Framework\Exception\NoSuchEntityException
      */
     public function getBulkShortStatus($bulkUuid);

app/code/Magento/AsynchronousOperations/Api/Data/BulkStatusInterface.php renamed to app/code/Magento/AsynchronousOperations/Api/Data/BulkOperationsStatusInterface.php

@@ -15,22 +15,22 @@
  *
  * @api
  */
-interface BulkStatusInterface extends BulkSummaryInterface
+interface BulkOperationsStatusInterface extends BulkSummaryInterface
 {
 
     const OPERATIONS_LIST = 'operations_list';
 
     /**
      * Retrieve list of operation with statuses (short data).
      *
-     * @return \Magento\AsynchronousOperations\Api\Data\OperationStatusInterface[]
+     * @return \Magento\AsynchronousOperations\Api\Data\SummaryOperationStatusInterface[]
      */
     public function getOperationsList();
 
     /**
      * Set operations list.
      *
-     * @param \Magento\AsynchronousOperations\Api\Data\OperationStatusInterface[] $operationStatusList
+     * @param \Magento\AsynchronousOperations\Api\Data\SummaryOperationStatusInterface[] $operationStatusList
      * @return $this
      */
     public function setOperationsList($operationStatusList);

app/code/Magento/AsynchronousOperations/Api/Data/DetailedBulkStatusInterface.php renamed to app/code/Magento/AsynchronousOperations/Api/Data/DetailedBulkOperationsStatusInterface.php

@@ -15,22 +15,22 @@
  *
  * @api
  */
-interface DetailedBulkStatusInterface extends BulkSummaryInterface
+interface DetailedBulkOperationsStatusInterface extends BulkSummaryInterface
 {
 
     const OPERATIONS_LIST = 'operations_list';
 
     /**
      * Retrieve operations list.
      *
-     * @return \Magento\AsynchronousOperations\Api\Data\OperationInterface[]
+     * @return \Magento\AsynchronousOperations\Api\Data\DetailedOperationStatusInterface[]
      */
     public function getOperationsList();
 
     /**
      * Set operations list.
      *
-     * @param \Magento\AsynchronousOperations\Api\Data\OperationInterface[] $operationStatusList
+     * @param \Magento\AsynchronousOperations\Api\Data\DetailedOperationStatusInterface[] $operationStatusList
      * @return $this
      */
     public function setOperationsList($operationStatusList);

app/code/Magento/AsynchronousOperations/Api/Data/DetailedOperationStatusInterface.php

@@ -0,0 +1,31 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+
+declare(strict_types=1);
+
+namespace Magento\AsynchronousOperations\Api\Data;
+
+/**
+ * @api
+ */
+interface DetailedOperationStatusInterface extends OperationInterface
+{
+    /**
+     * Result serialized Data
+     *
+     * @return string
+     */
+    public function getResultSerializedData();
+
+    /**
+     * Set result serialized data
+     *
+     * @param string $resultSerializedData
+     * @return $this
+     */
+    public function setResultSerializedData($resultSerializedData);
+}
+

app/code/Magento/AsynchronousOperations/Api/Data/OperationStatusInterface.php renamed to app/code/Magento/AsynchronousOperations/Api/Data/SummaryOperationStatusInterface.php

@@ -15,9 +15,8 @@
  * without serialized_data and result_serialized_data
  *
  * @api
- * @see \Magento\AsynchronousOperations\Api\Data\OperationInterface
  */
-interface OperationStatusInterface
+interface SummaryOperationStatusInterface
 {
     /**
      * Operation id

app/code/Magento/AsynchronousOperations/Model/BulkOperationsStatus.php

@@ -10,8 +10,8 @@
 
 use Magento\Framework\Exception\NoSuchEntityException;
 use Magento\Framework\EntityManager\EntityManager;
-use Magento\AsynchronousOperations\Api\Data\BulkStatusInterfaceFactory as BulkStatusShortFactory;
-use Magento\AsynchronousOperations\Api\Data\DetailedBulkStatusInterfaceFactory as BulkStatusDetailedFactory;
+use Magento\AsynchronousOperations\Api\Data\BulkOperationsStatusInterfaceFactory as BulkStatusShortFactory;
+use Magento\AsynchronousOperations\Api\Data\DetailedBulkOperationsStatusInterfaceFactory as BulkStatusDetailedFactory;
 use Magento\AsynchronousOperations\Api\Data\OperationDetailsInterfaceFactory;
 use Magento\AsynchronousOperations\Api\BulkStatusInterface;
 use Magento\AsynchronousOperations\Model\ResourceModel\Operation\CollectionFactory;
@@ -42,7 +42,7 @@ class BulkOperationsStatus implements BulkStatusInterface
     private $bulkStatus;
 
     /**
-     * @var \Magento\AsynchronousOperations\Api\Data\OperationInterfaceFactory
+     * @var CollectionFactory
      */
     private $operationCollectionFactory;
 
@@ -108,7 +108,7 @@ public function getBulkDetailedStatus($bulkUuid)
     {
         $bulkSummary = $this->bulkDetailedFactory->create();
 
-        /** @var \Magento\AsynchronousOperations\Api\Data\DetailedBulkStatusInterface $bulk */
+        /** @var \Magento\AsynchronousOperations\Api\Data\DetailedBulkOperationsStatusInterface $bulk */
         $bulk = $this->entityManager->load($bulkSummary, $bulkUuid);
 
         if ($bulk->getBulkId() === null) {
@@ -132,7 +132,7 @@ public function getBulkShortStatus($bulkUuid)
     {
         $bulkSummary = $this->bulkShortFactory->create();
 
-        /** @var \Magento\AsynchronousOperations\Api\Data\BulkStatusInterface $bulk */
+        /** @var \Magento\AsynchronousOperations\Api\Data\BulkOperationsStatusInterface $bulk */
         $bulk = $this->entityManager->load($bulkSummary, $bulkUuid);
         if ($bulk->getBulkId() === null) {
             throw new NoSuchEntityException(

app/code/Magento/AsynchronousOperations/Model/BulkStatus/CalculatedStatusSql.php

@@ -6,7 +6,6 @@
 
 namespace Magento\AsynchronousOperations\Model\BulkStatus;
 
-use Magento\AsynchronousOperations\Api\Data\OperationInterface;
 use Magento\AsynchronousOperations\Api\Data\BulkSummaryInterface;
 
 class CalculatedStatusSql

app/code/Magento/AsynchronousOperations/Model/BulkStatus/Detailed.php

@@ -8,10 +8,10 @@
 
 namespace Magento\AsynchronousOperations\Model\BulkStatus;
 
-use Magento\AsynchronousOperations\Api\Data\DetailedBulkStatusInterface;
+use Magento\AsynchronousOperations\Api\Data\DetailedBulkOperationsStatusInterface;
 use Magento\AsynchronousOperations\Model\BulkSummary;
 
-class Detailed extends BulkSummary implements DetailedBulkStatusInterface
+class Detailed extends BulkSummary implements DetailedBulkOperationsStatusInterface
 {
     /**
      * @inheritDoc

app/code/Magento/AsynchronousOperations/Model/BulkStatus/Short.php

@@ -8,10 +8,10 @@
 
 namespace Magento\AsynchronousOperations\Model\BulkStatus;
 
-use Magento\AsynchronousOperations\Api\Data\BulkStatusInterface;
+use Magento\AsynchronousOperations\Api\Data\BulkOperationsStatusInterface;
 use Magento\AsynchronousOperations\Model\BulkSummary;
 
-class Short extends BulkSummary implements BulkStatusInterface
+class Short extends BulkSummary implements BulkOperationsStatusInterface
 {
     /**
      * @inheritDoc

app/code/Magento/AsynchronousOperations/Model/Operation.php

@@ -5,14 +5,14 @@
  */
 namespace Magento\AsynchronousOperations\Model;
 
-use Magento\AsynchronousOperations\Api\Data\OperationInterface;
+use Magento\AsynchronousOperations\Api\Data\DetailedOperationStatusInterface;
 use Magento\Framework\DataObject;
 use Magento\Framework\Api\ExtensibleDataInterface;
 
 /**
  * Class Operation
  */
-class Operation extends DataObject implements OperationInterface, ExtensibleDataInterface
+class Operation extends DataObject implements DetailedOperationStatusInterface, ExtensibleDataInterface
 {
     /**
      * @inheritDoc

app/code/Magento/AsynchronousOperations/Model/OperationManagement.php

@@ -6,7 +6,7 @@
 
 namespace Magento\AsynchronousOperations\Model;
 
-use Magento\AsynchronousOperations\Api\Data\OperationInterfaceFactory;
+use Magento\AsynchronousOperations\Api\Data\DetailedOperationStatusInterfaceFactory;
 use Magento\Framework\EntityManager\EntityManager;
 
 /**
@@ -20,7 +20,7 @@ class OperationManagement implements \Magento\Framework\Bulk\OperationManagement
     private $entityManager;
 
     /**
-     * @var OperationInterfaceFactory
+     * @var DetailedOperationStatusInterfaceFactory
      */
     private $operationFactory;
 
@@ -33,31 +33,39 @@ class OperationManagement implements \Magento\Framework\Bulk\OperationManagement
      * OperationManagement constructor.
      *
      * @param EntityManager $entityManager
-     * @param OperationInterfaceFactory $operationFactory
+     * @param DetailedOperationStatusInterfaceFactory $operationFactory
      * @param \Psr\Log\LoggerInterface $logger
      */
     public function __construct(
         EntityManager $entityManager,
-        OperationInterfaceFactory $operationFactory,
+        DetailedOperationStatusInterfaceFactory $operationFactory,
         \Psr\Log\LoggerInterface $logger
     ) {
         $this->entityManager = $entityManager;
-        $this->operationFactory= $operationFactory;
+        $this->operationFactory = $operationFactory;
         $this->logger = $logger;
     }
 
     /**
      * @inheritDoc
      */
-    public function changeOperationStatus($operationId, $status, $errorCode = null, $message = null, $data = null)
-    {
+    public function changeOperationStatus(
+        $operationId,
+        $status,
+        $errorCode = null,
+        $message = null,
+        $data = null,
+        $resultData = null
+    ) {
         try {
             $operationEntity = $this->operationFactory->create();
             $this->entityManager->load($operationEntity, $operationId);
             $operationEntity->setErrorCode($errorCode);
             $operationEntity->setStatus($status);
             $operationEntity->setResultMessage($message);
             $operationEntity->setSerializedData($data);
+            $operationEntity->setResultSerializedData($resultData);
+            $operationEntity->setResultSerializedData($resultData);
             $this->entityManager->save($operationEntity);
         } catch (\Exception $exception) {
             $this->logger->critical($exception->getMessage());

app/code/Magento/AsynchronousOperations/Model/OperationProcessor.php

@@ -22,6 +22,8 @@
 use Magento\Framework\DB\Adapter\ConnectionException;
 use Magento\Framework\DB\Adapter\DeadlockException;
 use Magento\Framework\DB\Adapter\LockWaitException;
+use Magento\Framework\Webapi\ServiceOutputProcessor;
+use Magento\Framework\Communication\ConfigInterface as CommunicationConfig;
 
 /**
  * Class OperationProcessor
@@ -60,6 +62,16 @@ class OperationProcessor
      */
     private $logger;
 
+    /**
+     * @var ServiceOutputProcessor
+     */
+    private $serviceOutputProcessor;
+
+    /**
+     * @var CommunicationConfig
+     */
+    private $communicationConfig;
+
     /**
      * OperationProcessor constructor.
      *
@@ -69,13 +81,17 @@ class OperationProcessor
      * @param Json $jsonHelper
      * @param OperationManagementInterface $operationManagement
      * @param LoggerInterface $logger
+     * @param \Magento\Framework\Webapi\ServiceOutputProcessor $serviceOutputProcessor
+     * @param \Magento\Framework\Communication\ConfigInterface $communicationConfig
      */
     public function __construct(
         MessageValidator $messageValidator,
         MessageEncoder $messageEncoder,
         ConsumerConfigurationInterface $configuration,
         Json $jsonHelper,
         OperationManagementInterface $operationManagement,
+        ServiceOutputProcessor $serviceOutputProcessor,
+        CommunicationConfig $communicationConfig,
         LoggerInterface $logger
     ) {
         $this->messageValidator = $messageValidator;
@@ -84,6 +100,8 @@ public function __construct(
         $this->jsonHelper = $jsonHelper;
         $this->operationManagement = $operationManagement;
         $this->logger = $logger;
+        $this->serviceOutputProcessor = $serviceOutputProcessor;
+        $this->communicationConfig = $communicationConfig;
     }
 
     /**
@@ -113,12 +131,32 @@ public function process(string $encodedMessage)
             $messages[] = $e->getMessage();
         }
 
+        $outputData = null;
         if ($errorCode === null) {
             foreach ($handlers as $callback) {
                 $result = $this->executeHandler($callback, $entityParams);
                 $status = $result['status'];
                 $errorCode = $result['error_code'];
                 $messages = array_merge($messages, $result['messages']);
+                $outputData = $result['output_data'];
+            }
+        }
+
+        if (isset($outputData)) {
+            try {
+                $communicationConfig = $this->communicationConfig->getTopic($topicName);
+                $asyncHandler =
+                    $communicationConfig[CommunicationConfig::TOPIC_HANDLERS][AsyncConfig::DEFAULT_HANDLER_NAME];
+                $serviceClass = $asyncHandler[CommunicationConfig::HANDLER_TYPE];
+                $serviceMethod = $asyncHandler[CommunicationConfig::HANDLER_METHOD];
+                $outputData = $this->serviceOutputProcessor->process(
+                    $outputData,
+                    $serviceClass,
+                    $serviceMethod
+                );
+                $outputData = $this->jsonHelper->serialize($outputData);
+            } catch (\Exception $e) {
+                $messages[] = $e->getMessage();
             }
         }
 
@@ -128,7 +166,8 @@ public function process(string $encodedMessage)
             $status,
             $errorCode,
             implode('; ', $messages),
-            $serializedData
+            $serializedData,
+            $outputData
         );
     }
 
@@ -144,10 +183,11 @@ private function executeHandler($callback, $entityParams)
         $result = [
             'status' => OperationInterface::STATUS_TYPE_COMPLETE,
             'error_code' => null,
-            'messages' => []
+            'messages' => [],
+            'output_data' => null
         ];
         try {
-            call_user_func_array($callback, $entityParams);
+            $result['output_data'] = call_user_func_array($callback, $entityParams);
             $result['messages'][] = sprintf('Service execution success %s::%s', get_class($callback[0]), $callback[1]);
         } catch (\Zend_Db_Adapter_Exception  $e) {
             $this->logger->critical($e->getMessage());

app/code/Magento/AsynchronousOperations/Model/OperationStatus.php

@@ -9,14 +9,14 @@
 namespace Magento\AsynchronousOperations\Model;
 
 use Magento\AsynchronousOperations\Api\Data\OperationInterface;
-use Magento\AsynchronousOperations\Api\Data\OperationStatusInterface;
+use Magento\AsynchronousOperations\Api\Data\SummaryOperationStatusInterface;
 use Magento\Framework\DataObject;
 use Magento\Framework\Api\ExtensibleDataInterface;
 
 /**
  * Class OperationShortDetails
  */
-class OperationStatus extends DataObject implements OperationStatusInterface, ExtensibleDataInterface
+class OperationStatus extends DataObject implements SummaryOperationStatusInterface, ExtensibleDataInterface
 {
     /**
      * @inheritDoc