Merge pull request #58 from aligent/feature/remove-batchdatamapper-hack

Remove plugin hack

Model/Indexer/AsyncEventSubscriber.php

@@ -9,6 +9,7 @@
 use Aligent\AsyncEvents\Model\Resolver\AsyncEvent;
 use ArrayIterator;
 use Magento\CatalogSearch\Model\Indexer\IndexerHandlerFactory;
+use Magento\Elasticsearch\Model\Adapter\Elasticsearch;
 use Magento\Framework\App\DeploymentConfig;
 use Magento\Framework\App\ObjectManager;
 use Magento\Framework\Indexer\ActionInterface as IndexerActionInterface;
@@ -58,6 +59,7 @@ class AsyncEventSubscriber implements
      * @param AsyncEvent $asyncEventScopeResolver
      * @param IndexStructure $indexStructure
      * @param Config $config
+     * @param Elasticsearch $dataMapperAdapter
      * @param array $data
      * @param int|null $batchSize
      * @param DeploymentConfig|null $deploymentConfig
@@ -69,6 +71,7 @@ public function __construct(
         private readonly AsyncEvent $asyncEventScopeResolver,
         private readonly IndexStructureInterface $indexStructure,
         private readonly Config $config,
+        private readonly Elasticsearch $dataMapperAdapter,
         private readonly array $data,
         int $batchSize = null,
         DeploymentConfig $deploymentConfig = null
@@ -99,6 +102,7 @@ public function executeByDimensions(array $dimensions, ?Traversable $entityIds)
         $saveHandler = $this->indexerHandlerFactory->create(
             [
                 'data' => $this->data,
+                'adapter' => $this->dataMapperAdapter,
                 'scopeResolver' => $this->asyncEventScopeResolver,
                 'indexStructure' => $this->indexStructure
             ]

Test/Integration/EventRetryTest.php

@@ -0,0 +1,108 @@
+<?php
+
+namespace Aligent\AsyncEvents\Test\Integration;
+
+use Aligent\AsyncEvents\Helper\QueueMetadataInterface;
+use Aligent\AsyncEvents\Model\Config;
+use Magento\Framework\App\ResourceConnection;
+use Magento\Framework\MessageQueue\PublisherInterface;
+use Magento\Framework\Serialize\Serializer\Json;
+use Magento\TestFramework\Helper\Amqp;
+use Magento\TestFramework\Helper\Bootstrap;
+use Magento\TestFramework\MessageQueue\EnvironmentPreconditionException;
+use Magento\TestFramework\MessageQueue\PreconditionFailedException;
+use Magento\TestFramework\MessageQueue\PublisherConsumerController;
+use PHPUnit\Framework\TestCase;
+
+class EventRetryTest extends TestCase
+{
+    /** @var Amqp|null */
+    private ?Amqp $helper;
+
+    /** @var PublisherInterface|null */
+    private ?PublisherInterface $publisher;
+
+    /** @var PublisherConsumerController|null */
+    private ?PublisherConsumerController $publisherConsumerController;
+
+    /** @var Json|null */
+    private ?Json $json;
+
+    protected function setUp(): void
+    {
+        Bootstrap::getObjectManager()->configure([
+            'preferences' => [
+                Config::class => TestConfig::class
+            ]
+        ]);
+
+        $this->helper = Bootstrap::getObjectManager()->create(Amqp::class);
+        $this->publisher = Bootstrap::getObjectManager()->create(PublisherInterface::class);
+        $this->json = Bootstrap::getObjectManager()->create(Json::class);
+        $this->connection = Bootstrap::getObjectManager()->get(ResourceConnection::class);
+
+        if (!$this->helper->isAvailable()) {
+            $this->fail('This test relies on RabbitMQ Management Plugin.');
+        }
+    }
+
+    /**
+     * Test event retries
+     *
+     * Disable database isolation because the transactions need to be committed so that the consumer can
+     * retrieve the subscriptions from database in a separate consumer process.
+     *
+     * @magentoDataFixture Aligent_AsyncEvents::Test/_files/http_async_events.php
+     * @magentoDbIsolation disabled
+     * @magentoConfigFixture default/system/async_events/max_deaths 3
+     */
+    public function testRetry()
+    {
+        $this->publisher->publish(
+            QueueMetadataInterface::EVENT_QUEUE,
+            [
+                'example.event',
+                $this->json->serialize([
+                    'countryId' => 'AU'
+                ]),
+            ]
+        );
+
+        $this->publisherConsumerController = Bootstrap::getObjectManager()->create(
+            PublisherConsumerController::class,
+            [
+                'consumers' => ['event.trigger.consumer', 'event.retry.consumer'],
+                'logFilePath' => TESTS_TEMP_DIR . "/MessageQueueTestLog.txt",
+                'maxMessages' => 10,
+                'appInitParams' => Bootstrap::getInstance()->getAppInitParams()
+            ]
+        );
+
+        try {
+            $this->publisherConsumerController->startConsumers();
+            sleep(16);
+        } catch (EnvironmentPreconditionException $e) {
+            $this->markTestSkipped($e->getMessage());
+        } catch (PreconditionFailedException $e) {
+            $this->fail(
+                $e->getMessage()
+            );
+        } finally {
+            $this->publisherConsumerController->stopConsumers();
+        }
+
+        $table = $this->connection->getTableName('async_event_subscriber_log');
+        $connection = $this->connection->getConnection();
+        $select = $connection->select()
+            ->from($table, 'uuid')
+            ->columns(['events' => new \Zend_Db_Expr('COUNT(*)')])
+            ->group('uuid');
+
+        $events = $connection->fetchAll($select);
+
+        foreach ($events as $event) {
+            // An uuid batch should be retired for 3 times after the first attempt. 1 + 3
+            $this->assertEquals(4, $event['events']);
+        }
+    }
+}

Test/Integration/FailoverTopologyTest.php

@@ -9,7 +9,6 @@
 
 namespace Aligent\AsyncEvents\Test\Integration;
 
-use Aligent\AsyncEvents\Helper\QueueMetadataInterface;
 use Aligent\AsyncEvents\Service\AsyncEvent\RetryManager;
 use Magento\TestFramework\Helper\Amqp;
 use Magento\TestFramework\Helper\Bootstrap;

Test/Integration/TestConfig.php

@@ -0,0 +1,17 @@
+<?php
+
+namespace Aligent\AsyncEvents\Test\Integration;
+
+use Aligent\AsyncEvents\Model\Config;
+use Magento\Directory\Api\CountryInformationAcquirerInterface;
+
+class TestConfig extends Config
+{
+    public function get(string $key): array
+    {
+        return [
+            "class" => CountryInformationAcquirerInterface::class,
+            "method" => "getCountryInfo",
+        ];
+    }
+}

Test/_files/http_async_events.php

@@ -1,25 +1,19 @@
 <?php
 
-use Aligent\AsyncEvents\Api\Data\AsyncEventInterface;
-use Aligent\AsyncEvents\Api\Data\AsyncEventInterfaceFactory;
+use Magento\Framework\App\ResourceConnection;
 use Magento\TestFramework\Helper\Bootstrap;
 
 $objectManager = Bootstrap::getObjectManager();
 
-$asyncEventFactory = $objectManager->get(AsyncEventInterfaceFactory::class);
+$resource = $objectManager->get(ResourceConnection::class);
+$connection = $resource->getConnection();
 
-$asyncEventRepository = $objectManager->get(\Aligent\AsyncEvents\Api\AsyncEventRepositoryInterface::class);
-
-/** @var AsyncEventInterface $asyncEvent */
-$asyncEvent = $asyncEventFactory->create(
-    [
-        'data' => [
-            'event' => 'example.event',
-            'recipient_url' => 'http://host.docker.internal:3001/failable',
-            'verification_token' => 'supersecret',
-            'status' => 1
-        ]
-    ]
-);
-$asyncEvent->setEventName('example.event');
-$asyncEventRepository->save($asyncEvent);
+$connection->insertOnDuplicate('async_event_subscriber', [
+    'subscription_id' => 1,
+    'event_name' => 'example.event',
+    'recipient_url' => 'https://mock.codes/500',
+    'status' => 1,
+    'metadata' => 'http',
+    'verification_token' => 'secret',
+    'subscribed_at' => (new DateTime())->format(DateTimeInterface::ATOM)
+]);

etc/di.xml

@@ -94,6 +94,8 @@
     <type name="Aligent\AsyncEvents\Model\Indexer\AsyncEventSubscriber">
         <arguments>
             <argument name="dimensionProvider" xsi:type="object" shared="false">\Aligent\AsyncEvents\Model\Indexer\AsyncEventDimensionProvider</argument>
+            <argument name="dataMapperAdapter" xsi:type="object">dataMapperAdapter</argument>
+            <argument name="indexStructure" xsi:type="object">dataMapperIndexStructure</argument>
         </arguments>
     </type>
 
@@ -123,13 +125,22 @@
         </arguments>
     </type>
 
-    <type name="Magento\Elasticsearch\Model\Adapter\BatchDataMapperInterface">
-        <plugin name="add_entity_type_context"
-                type="Aligent\AsyncEvents\Plugin\AddEntityTypeContext"/>
-    </type>
-
     <type name="Magento\Elasticsearch\Model\Adapter\Index\BuilderInterface">
         <plugin name="map_string_scope_to_int"
                 type="Aligent\AsyncEvents\Plugin\MapStringScopeToInt"/>
     </type>
+
+    <virtualType name="dataMapperIndexStructure" type="Magento\Elasticsearch\Model\Indexer\IndexStructure">
+        <arguments>
+            <argument name="adapter" xsi:type="object">
+                dataMapperAdapter
+            </argument>
+        </arguments>
+    </virtualType>
+
+    <virtualType name="dataMapperAdapter" type="Magento\Elasticsearch\Model\Adapter\Elasticsearch">
+        <arguments>
+            <argument name="batchDocumentDataMapper" xsi:type="object">\Aligent\AsyncEvents\Model\Adapter\BatchDataMapper\AsyncEventLogMapper</argument>
+        </arguments>
+    </virtualType>
 </config>

etc/queue_consumer.xml

@@ -6,5 +6,7 @@
               handler="Aligent\AsyncEvents\Model\AsyncEventTriggerHandler::process"/>
 
     <consumer name="event.retry.consumer" queue="event.failover.retry" connection="amqp"
-              consumerInstance="Magento\Framework\MessageQueue\Consumer" />
+              consumerInstance="Magento\Framework\MessageQueue\Consumer"
+              handler="Aligent\AsyncEvents\Model\RetryHandler::process"
+    />
 </config>