ENGCOM-6171: Fix for Mysql MessageQueue (2) #25289

VladimirZaets · web-flow · commit bbb32196c617 · 2019-11-06T14:47:51.000-06:00
diff --git a/app/code/Magento/MysqlMq/Model/Driver/Bulk/Exchange.php b/app/code/Magento/MysqlMq/Model/Driver/Bulk/Exchange.php
@@ -3,17 +3,24 @@
  * Copyright © Magento, Inc. All rights reserved.
  * See COPYING.txt for license details.
  */
+
 namespace Magento\MysqlMq\Model\Driver\Bulk;
 
 use Magento\Framework\MessageQueue\Bulk\ExchangeInterface;
-use Magento\Framework\MessageQueue\ConfigInterface as MessageQueueConfig;
+use Magento\Framework\MessageQueue\Topology\ConfigInterface as MessageQueueConfig;
+use Magento\MysqlMq\Model\ConnectionTypeResolver;
 use Magento\MysqlMq\Model\QueueManagement;
 
 /**
  * Used to send messages in bulk in MySQL queue.
  */
 class Exchange implements ExchangeInterface
 {
+    /**
+     * @var ConnectionTypeResolver
+     */
+    private $connectionTypeResolver;
+
     /**
      * @var MessageQueueConfig
      */
@@ -27,21 +34,39 @@ class Exchange implements ExchangeInterface
     /**
      * Initialize dependencies.
      *
+     * @param ConnectionTypeResolver $connectionTypeResolver
      * @param MessageQueueConfig $messageQueueConfig
      * @param QueueManagement $queueManagement
      */
-    public function __construct(MessageQueueConfig $messageQueueConfig, QueueManagement $queueManagement)
-    {
+    public function __construct(
+        ConnectionTypeResolver $connectionTypeResolver,
+        MessageQueueConfig $messageQueueConfig,
+        QueueManagement $queueManagement
+    ) {
         $this->messageQueueConfig = $messageQueueConfig;
         $this->queueManagement = $queueManagement;
+        $this->connectionTypeResolver = $connectionTypeResolver;
     }
 
     /**
      * @inheritdoc
      */
     public function enqueue($topic, array $envelopes)
     {
-        $queueNames = $this->messageQueueConfig->getQueuesByTopic($topic);
+        $queueNames = [];
+        $exchanges = $this->messageQueueConfig->getExchanges();
+        foreach ($exchanges as $exchange) {
+            $connection = $exchange->getConnection();
+            if ($this->connectionTypeResolver->getConnectionType($connection)) {
+                foreach ($exchange->getBindings() as $binding) {
+                    // This only supports exact matching of topics.
+                    if ($binding->getTopic() === $topic) {
+                        $queueNames[] = $binding->getDestination();
+                    }
+                }
+            }
+        }
+
         $messages = array_map(
             function ($envelope) {
                 return $envelope->getBody();
diff --git a/app/code/Magento/MysqlMq/Model/Driver/Exchange.php b/app/code/Magento/MysqlMq/Model/Driver/Exchange.php
@@ -3,15 +3,25 @@
  * Copyright © Magento, Inc. All rights reserved.
  * See COPYING.txt for license details.
  */
+
 namespace Magento\MysqlMq\Model\Driver;
 
 use Magento\Framework\MessageQueue\EnvelopeInterface;
 use Magento\Framework\MessageQueue\ExchangeInterface;
-use Magento\Framework\MessageQueue\ConfigInterface as MessageQueueConfig;
+use Magento\Framework\MessageQueue\Topology\ConfigInterface as MessageQueueConfig;
+use Magento\MysqlMq\Model\ConnectionTypeResolver;
 use Magento\MysqlMq\Model\QueueManagement;
 
+/**
+ * Class Exchange
+ */
 class Exchange implements ExchangeInterface
 {
+    /**
+     * @var ConnectionTypeResolver
+     */
+    private $connectionTypeResolver;
+
     /**
      * @var MessageQueueConfig
      */
@@ -25,13 +35,18 @@ class Exchange implements ExchangeInterface
     /**
      * Initialize dependencies.
      *
+     * @param ConnectionTypeResolver $connectionTypeResolver
      * @param MessageQueueConfig $messageQueueConfig
      * @param QueueManagement $queueManagement
      */
-    public function __construct(MessageQueueConfig $messageQueueConfig, QueueManagement $queueManagement)
-    {
+    public function __construct(
+        ConnectionTypeResolver $connectionTypeResolver,
+        MessageQueueConfig $messageQueueConfig,
+        QueueManagement $queueManagement
+    ) {
         $this->messageQueueConfig = $messageQueueConfig;
         $this->queueManagement = $queueManagement;
+        $this->connectionTypeResolver = $connectionTypeResolver;
     }
 
     /**
@@ -43,7 +58,18 @@ public function __construct(MessageQueueConfig $messageQueueConfig, QueueManagem
      */
     public function enqueue($topic, EnvelopeInterface $envelope)
     {
-        $queueNames = $this->messageQueueConfig->getQueuesByTopic($topic);
+        $queueNames = [];
+        $exchanges = $this->messageQueueConfig->getExchanges();
+        foreach ($exchanges as $exchange) {
+            $connection = $exchange->getConnection();
+            if ($this->connectionTypeResolver->getConnectionType($connection)) {
+                foreach ($exchange->getBindings() as $binding) {
+                    if ($binding->getTopic() == $topic) {
+                        $queueNames[] = $binding->getDestination();
+                    }
+                }
+            }
+        }
         $this->queueManagement->addMessageToQueues($topic, $envelope->getBody(), $queueNames);
         return null;
     }
diff --git a/app/code/Magento/MysqlMq/Setup/Recurring.php b/app/code/Magento/MysqlMq/Setup/Recurring.php
@@ -3,12 +3,13 @@
  * Copyright © Magento, Inc. All rights reserved.
  * See COPYING.txt for license details.
  */
+
 namespace Magento\MysqlMq\Setup;
 
 use Magento\Framework\Setup\InstallSchemaInterface;
 use Magento\Framework\Setup\ModuleContextInterface;
 use Magento\Framework\Setup\SchemaSetupInterface;
-use Magento\Framework\MessageQueue\ConfigInterface as MessageQueueConfig;
+use Magento\Framework\MessageQueue\Topology\ConfigInterface as MessageQueueConfig;
 
 /**
  * Class Recurring
@@ -29,17 +30,17 @@ public function __construct(MessageQueueConfig $messageQueueConfig)
     }
 
     /**
-     * {@inheritdoc}
+     * @inheritdoc
      */
     public function install(SchemaSetupInterface $setup, ModuleContextInterface $context)
     {
         $setup->startSetup();
 
-        $binds = $this->messageQueueConfig->getBinds();
         $queues = [];
-        foreach ($binds as $bind) {
-            $queues[] = $bind[MessageQueueConfig::BIND_QUEUE];
+        foreach ($this->messageQueueConfig->getQueues() as $queue) {
+            $queues[] = $queue->getName();
         }
+
         $connection = $setup->getConnection();
         $existingQueues = $connection->fetchCol($connection->select()->from($setup->getTable('queue'), 'name'));
         $queues = array_unique(array_diff($queues, $existingQueues));
diff --git a/app/code/Magento/MysqlMq/Test/Unit/Model/Driver/Bulk/ExchangeTest.php b/app/code/Magento/MysqlMq/Test/Unit/Model/Driver/Bulk/ExchangeTest.php
@@ -25,6 +25,10 @@ class ExchangeTest extends \PHPUnit\Framework\TestCase
      * @var \Magento\MysqlMq\Model\Driver\Bulk\Exchange
      */
     private $exchange;
+    /**
+     * @var \Magento\MysqlMq\Model\ConnectionTypeResolver|\PHPUnit_Framework_MockObject_MockObject
+     */
+    private $connnectionTypeResolver;
 
     /**
      * Set up.
@@ -33,15 +37,20 @@ class ExchangeTest extends \PHPUnit\Framework\TestCase
      */
     protected function setUp()
     {
-        $this->messageQueueConfig = $this->getMockBuilder(\Magento\Framework\MessageQueue\ConfigInterface::class)
+        $this->messageQueueConfig = $this->getMockBuilder(
+            \Magento\Framework\MessageQueue\Topology\ConfigInterface::class
+        )
             ->disableOriginalConstructor()->getMock();
         $this->queueManagement = $this->getMockBuilder(\Magento\MysqlMq\Model\QueueManagement::class)
             ->disableOriginalConstructor()->getMock();
+        $this->connnectionTypeResolver = $this->getMockBuilder(\Magento\MysqlMq\Model\ConnectionTypeResolver::class)
+            ->disableOriginalConstructor()->getMock();
 
         $objectManager = new \Magento\Framework\TestFramework\Unit\Helper\ObjectManager($this);
         $this->exchange = $objectManager->getObject(
             \Magento\MysqlMq\Model\Driver\Bulk\Exchange::class,
             [
+                'connectionTypeResolver' => $this->connnectionTypeResolver,
                 'messageQueueConfig' => $this->messageQueueConfig,
                 'queueManagement' => $this->queueManagement,
             ]
@@ -56,10 +65,46 @@ protected function setUp()
     public function testEnqueue()
     {
         $topicName = 'topic.name';
-        $queueNames = ['queue0', 'queue1'];
+        $queueNames = ['queue0'];
+        $binding1 = $this->createMock(
+            \Magento\Framework\MessageQueue\Topology\Config\ExchangeConfigItem\BindingInterface::class
+        );
+        $binding1->expects($this->once())
+            ->method('getTopic')
+            ->willReturn($topicName);
+        $binding1->expects($this->once())
+            ->method('getDestination')
+            ->willReturn($queueNames[0]);
+        $binding2 = $this->createMock(
+            \Magento\Framework\MessageQueue\Topology\Config\ExchangeConfigItem\BindingInterface::class
+        );
+        $binding2->expects($this->once())
+            ->method('getTopic')
+            ->willReturn('different.topic');
+        $binding2->expects($this->never())
+            ->method('getDestination');
+        $exchange1 = $this->createMock(
+            \Magento\Framework\MessageQueue\Topology\Config\ExchangeConfigItemInterface::class
+        );
+        $exchange1->expects($this->once())
+            ->method('getConnection')
+            ->willReturn('db');
+        $exchange1->expects($this->once())
+            ->method('getBindings')
+            ->willReturn([$binding1, $binding2]);
+        $exchange2 = $this->createMock(
+            \Magento\Framework\MessageQueue\Topology\Config\ExchangeConfigItemInterface::class
+        );
+        $exchange2->expects($this->once())
+            ->method('getConnection')
+            ->willReturn('amqp');
+        $exchange2->expects($this->never())
+            ->method('getBindings');
+
+        $this->connnectionTypeResolver->method('getConnectionType')->willReturnOnConsecutiveCalls(['db', null]);
         $envelopeBody = 'serializedMessage';
         $this->messageQueueConfig->expects($this->once())
-            ->method('getQueuesByTopic')->with($topicName)->willReturn($queueNames);
+            ->method('getExchanges')->willReturn([$exchange1, $exchange2]);
         $envelope = $this->getMockBuilder(\Magento\Framework\MessageQueue\EnvelopeInterface::class)
             ->disableOriginalConstructor()->getMock();
         $envelope->expects($this->once())->method('getBody')->willReturn($envelopeBody);
diff --git a/app/code/Magento/MysqlMq/Test/Unit/Setup/RecurringTest.php b/app/code/Magento/MysqlMq/Test/Unit/Setup/RecurringTest.php
@@ -34,7 +34,9 @@ class RecurringTest extends \PHPUnit\Framework\TestCase
     protected function setUp()
     {
         $this->objectManager = new ObjectManager($this);
-        $this->messageQueueConfig = $this->getMockBuilder(\Magento\Framework\MessageQueue\ConfigInterface::class)
+        $this->messageQueueConfig = $this->getMockBuilder(
+            \Magento\Framework\MessageQueue\Topology\ConfigInterface::class
+        )
             ->getMockForAbstractClass();
         $this->model = $this->objectManager->getObject(
             \Magento\MysqlMq\Setup\Recurring::class,
@@ -49,23 +51,14 @@ protected function setUp()
      */
     public function testInstall()
     {
-        $binds = [
-            'first_bind' => [
-                'queue' => 'queue_name_1',
-                'exchange' => 'magento-db',
-                'topic' => 'queue.topic.1'
-            ],
-            'second_bind' => [
-                'queue' => 'queue_name_2',
-                'exchange' => 'magento-db',
-                'topic' => 'queue.topic.2'
-            ],
-            'third_bind' => [
-                'queue' => 'queue_name_3',
-                'exchange' => 'magento-db',
-                'topic' => 'queue.topic.3'
-            ]
-        ];
+        for ($i = 1; $i <= 3; $i++) {
+            $queue = $this->createMock(\Magento\Framework\MessageQueue\Topology\Config\QueueConfigItemInterface::class);
+            $queue->expects($this->once())
+                ->method('getName')
+                ->willReturn('queue_name_' . $i);
+            $queues[] = $queue;
+        }
+
         $dbQueues = [
             'queue_name_1',
             'queue_name_2',
@@ -81,7 +74,7 @@ public function testInstall()
             ->getMockForAbstractClass();
 
         $setup->expects($this->once())->method('startSetup')->willReturnSelf();
-        $this->messageQueueConfig->expects($this->once())->method('getBinds')->willReturn($binds);
+        $this->messageQueueConfig->expects($this->once())->method('getQueues')->willReturn($queues);
         $connection = $this->getMockBuilder(\Magento\Framework\DB\Adapter\AdapterInterface::class)
             ->getMockForAbstractClass();
         $setup->expects($this->once())->method('getConnection')->willReturn($connection);
