Merge branch 'B2B-2176' into foxes-pr

Author: dhaecker

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

@@ -79,7 +79,7 @@ public function __construct(
     }
 
     /**
-     * {@inheritdoc}
+     * @inheritdoc
      */
     public function publish($topicName, $data)
     {
@@ -91,6 +91,7 @@ public function publish($topicName, $data)
                 [
                     'body' => $message,
                     'properties' => [
+                        'topic_name' => $topicName,
                         'delivery_mode' => 2,
                         'message_id' => $this->messageIdGenerator->generate($topicName),
                     ]

app/code/Magento/MysqlMq/Model/Driver/Bulk/Exchange.php

@@ -3,10 +3,12 @@
  * Copyright © Magento, Inc. All rights reserved.
  * See COPYING.txt for license details.
  */
+declare(strict_types=1);
 
 namespace Magento\MysqlMq\Model\Driver\Bulk;
 
 use Magento\Framework\MessageQueue\Bulk\ExchangeInterface;
+use Magento\Framework\MessageQueue\Topology\Config\ExchangeConfigItem\BindingInterface;
 use Magento\Framework\MessageQueue\Topology\ConfigInterface as MessageQueueConfig;
 use Magento\MysqlMq\Model\ConnectionTypeResolver;
 use Magento\MysqlMq\Model\QueueManagement;
@@ -59,8 +61,7 @@ public function enqueue($topic, array $envelopes)
             $connection = $exchange->getConnection();
             if ($this->connectionTypeResolver->getConnectionType($connection)) {
                 foreach ($exchange->getBindings() as $binding) {
-                    // This only supports exact matching of topics.
-                    if ($binding->getTopic() === $topic) {
+                    if ($this->isMatchedBinding($binding, $topic)) {
                         $queueNames[] = $binding->getDestination();
                     }
                 }
@@ -77,4 +78,17 @@ function ($envelope) {
 
         return null;
     }
+
+    /**
+     * Check if the binding is matched by the topic
+     *
+     * @param BindingInterface $binding
+     * @param string $topic
+     * @return bool
+     */
+    private function isMatchedBinding(BindingInterface $binding, string $topic): bool
+    {
+        $pattern = '/^' . str_replace(['.', '*', '#'], ['\.', '[^.]+?', '(.*?)'], $binding->getTopic()) . '$/';
+        return preg_match($pattern, $topic) ? true : false;
+    }
 }

app/code/Magento/WebapiAsync/Code/Generator/Config/RemoteServiceReader/Publisher.php

@@ -9,6 +9,8 @@
 namespace Magento\WebapiAsync\Code\Generator\Config\RemoteServiceReader;
 
 use Magento\AsynchronousOperations\Model\ConfigInterface as WebApiAsyncConfig;
+use Magento\Framework\App\ObjectManager;
+use Magento\Framework\MessageQueue\DefaultValueProvider;
 
 /**
  * Remote service reader with auto generated configuration for queue_publisher.xml
@@ -20,15 +22,24 @@ class Publisher implements \Magento\Framework\Config\ReaderInterface
      */
     private $webapiAsyncConfig;
 
+    /**
+     * @var DefaultValueProvider
+     */
+    private $defaultValueProvider;
+
     /**
      * Initialize dependencies.
      *
      * @param WebApiAsyncConfig $webapiAsyncConfig
+     * @param DefaultValueProvider|null $defaultValueProvider
      */
     public function __construct(
-        WebApiAsyncConfig $webapiAsyncConfig
+        WebApiAsyncConfig $webapiAsyncConfig,
+        DefaultValueProvider $defaultValueProvider = null
     ) {
         $this->webapiAsyncConfig = $webapiAsyncConfig;
+        $this->defaultValueProvider = $defaultValueProvider
+            ?? ObjectManager::getInstance()->get(DefaultValueProvider::class);
     }
 
     /**
@@ -49,8 +60,8 @@ public function read($scope = null)
                     'topic'       => $topicName,
                     'disabled'    => false,
                     'connections' => [
-                        'amqp' => [
-                            'name'     => 'amqp',
+                        $this->defaultValueProvider->getConnection() => [
+                            'name'     => $this->defaultValueProvider->getConnection(),
                             'exchange' => 'magento',
                             'disabled' => false,
                         ],

app/code/Magento/WebapiAsync/etc/queue_consumer.xml

@@ -7,6 +7,6 @@
 -->
 <config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/consumer.xsd">
-    <consumer name="async.operations.all" queue="async.operations.all" connection="amqp"
+    <consumer name="async.operations.all" queue="async.operations.all"
               consumerInstance="Magento\AsynchronousOperations\Model\MassConsumer"/>
 </config>

app/code/Magento/WebapiAsync/etc/queue_topology.xml

@@ -6,7 +6,7 @@
  */
 -->
 <config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/topology.xsd">
-    <exchange name="magento" connection="amqp">
+    <exchange name="magento">
         <binding id="async.operations.all" topic="async.#" destination="async.operations.all"/>
     </exchange>
 </config>

dev/tests/integration/etc/post-install-setup-command-config.php.dist

@@ -16,5 +16,9 @@ return [
             '--remote-storage-region' => 'us-east-1'
             */
         ]
+    ],
+    [
+        'command' => 'setup:upgrade',
+        'config' => []
     ]
 ];

dev/tests/integration/framework/Magento/TestFramework/MessageQueue/ClearQueueProcessor.php

@@ -0,0 +1,69 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+declare(strict_types=1);
+
+namespace Magento\TestFramework\MessageQueue;
+
+use Magento\Framework\Exception\LocalizedException;
+use Magento\Framework\MessageQueue\Consumer\Config\ConsumerConfigItemInterface;
+use Magento\Framework\MessageQueue\Consumer\ConfigInterface as ConsumerConfig;
+use Magento\Framework\MessageQueue\ConsumerFactory;
+use Magento\Framework\MessageQueue\QueueRepository;
+
+/**
+ * The processor to clear message queue
+ */
+class ClearQueueProcessor
+{
+    /**
+     * @var ConsumerConfig
+     */
+    private $consumerConfig;
+
+    /**
+     * @var ConsumerFactory
+     */
+    private $consumerFactory;
+
+    /**
+     * @var QueueRepository
+     */
+    private $queueRepository;
+
+    /**
+     * ClearQueueProcessor constructor.
+     *
+     * @param ConsumerConfig $consumerConfig
+     * @param ConsumerFactory $consumerFactory
+     * @param QueueRepository $queueRepository
+     */
+    public function __construct(
+        ConsumerConfig $consumerConfig,
+        ConsumerFactory $consumerFactory,
+        QueueRepository $queueRepository
+    ) {
+        $this->consumerConfig = $consumerConfig;
+        $this->consumerFactory = $consumerFactory;
+        $this->queueRepository = $queueRepository;
+    }
+
+    /**
+     * Clear queue
+     *
+     * @param string $consumerName
+     * @throws LocalizedException
+     * return void
+     */
+    public function execute(string $consumerName): void
+    {
+        /** @var ConsumerConfigItemInterface $consumerConfig */
+        $consumerConfig = $this->consumerConfig->getConsumer($consumerName);
+        $queue = $this->queueRepository->get($consumerConfig->getConnection(), $consumerConfig->getQueue());
+        while ($message = $queue->dequeue()) {
+            $queue->acknowledge($message);
+        }
+    }
+}

dev/tests/integration/framework/Magento/TestFramework/MessageQueue/PublisherConsumerController.php

@@ -10,11 +10,9 @@
 
 use Magento\Framework\MessageQueue\PublisherInterface;
 use Magento\Framework\OsInfo;
+use Magento\TestFramework\Helper\Bootstrap;
 use Magento\TestFramework\Helper\Amqp;
 
-/**
- * Publisher Consumer Controller
- */
 class PublisherConsumerController
 {
     /**
@@ -52,6 +50,11 @@ class PublisherConsumerController
      */
     private $amqpHelper;
 
+    /**
+     * @var ClearQueueProcessor
+     */
+    private $clearQueueProcessor;
+
     /**
      * PublisherConsumerController constructor.
      * @param PublisherInterface $publisher
@@ -61,6 +64,7 @@ class PublisherConsumerController
      * @param array $consumers
      * @param array $appInitParams
      * @param null|int $maxMessages
+     * @param ClearQueueProcessor $clearQueueProcessor
      */
     public function __construct(
         PublisherInterface $publisher,
@@ -69,7 +73,8 @@ public function __construct(
         $logFilePath,
         $consumers,
         $appInitParams,
-        $maxMessages = null
+        $maxMessages = null,
+        ClearQueueProcessor $clearQueueProcessor = null
     ) {
         $this->consumers = $consumers;
         $this->publisher = $publisher;
@@ -78,6 +83,8 @@ public function __construct(
         $this->osInfo = $osInfo;
         $this->appInitParams = $appInitParams;
         $this->amqpHelper = $amqpHelper;
+        $this->clearQueueProcessor = $clearQueueProcessor
+            ?: Bootstrap::getObjectManager()->get(ClearQueueProcessor::class);
     }
 
     /**
@@ -90,12 +97,7 @@ public function initialize()
     {
         $this->validateEnvironmentPreconditions();
 
-        $connections = $this->amqpHelper->getConnections();
-        foreach (array_keys($connections) as $connectionName) {
-            $this->amqpHelper->deleteConnection($connectionName);
-        }
-        $this->amqpHelper->clearQueue("async.operations.all");
-
+        $this->clearQueueProcessor->execute("async.operations.all");
         $this->stopConsumers();
         $this->startConsumers();
 
@@ -123,12 +125,6 @@ private function validateEnvironmentPreconditions()
                 "This test relies on *nix shell and should be skipped in Windows environment."
             );
         }
-
-        if (!$this->amqpHelper->isAvailable()) {
-            throw new PreconditionFailedException(
-                'This test relies on RabbitMQ Management Plugin.'
-            );
-        }
     }
 
     /**