B2B-2171: Enable WebapiAsync work with redis connection

Author: joanhe

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

@@ -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/Consumer.php

@@ -60,8 +60,8 @@ public function read($scope = null)
                     'topic'       => $topicName,
                     'disabled'    => false,
                     'connections' => [
-                        $this->getConnection() => [
-                            'name'     => $this->getConnection(),
+                        $this->defaultValueProvider->getConnection() => [
+                            'name'     => $this->defaultValueProvider->getConnection(),
                             'exchange' => 'magento',
                             'disabled' => false,
                         ],
@@ -71,16 +71,4 @@ public function read($scope = null)
 
         return $result;
     }
-
-    /**
-     * Get connection
-     *
-     * @return string
-     */
-    private function getConnection()
-    {
-        $connection = $this->defaultValueProvider->getConnection();
-        // if db connection, return amqp instead.
-        return $connection === 'db' ? WebApiAsyncConfig::DEFAULT_CONSUMER_CONNECTION : $connection;
-    }
 }

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

@@ -0,0 +1,12 @@
+<?xml version="1.0"?>
+<!--
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+-->
+<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"
+              consumerInstance="Magento\AsynchronousOperations\Model\MassConsumer"/>
+</config>

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

@@ -0,0 +1,12 @@
+<?xml version="1.0"?>
+<!--
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+-->
+<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/topology.xsd">
+    <exchange name="magento">
+        <binding id="async.operations.all" topic="async.#" destination="async.operations.all"/>
+    </exchange>
+</config>

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

@@ -1682,7 +1682,6 @@
     <type name="Magento\Framework\MessageQueue\Consumer\Config\CompositeReader">
         <arguments>
             <argument name="readers" xsi:type="array">
-                <item name="asyncServiceReader" xsi:type="object" sortOrder="0">Magento\WebapiAsync\Code\Generator\Config\RemoteServiceReader\Consumer</item>
                 <item name="xmlReader" xsi:type="object" sortOrder="10">Magento\Framework\MessageQueue\Consumer\Config\Xml\Reader</item>
                 <item name="envReader" xsi:type="object" sortOrder="20">Magento\Framework\MessageQueue\Consumer\Config\Env\Reader</item>
             </argument>
@@ -1728,7 +1727,6 @@
     <type name="Magento\Framework\MessageQueue\Topology\Config\CompositeReader">
         <arguments>
             <argument name="readers" xsi:type="array">
-                <item name="asyncServiceReader" xsi:type="object" sortOrder="0">Magento\WebapiAsync\Code\Generator\Config\RemoteServiceReader\Topology</item>
                 <item name="remoteServiceReader" xsi:type="object" sortOrder="10">Magento\Framework\MessageQueue\Topology\Config\RemoteService\Reader</item>
                 <item name="xmlReader" xsi:type="object" sortOrder="20">Magento\Framework\MessageQueue\Topology\Config\Xml\Reader</item>
             </argument>