B2B-2171: Enable WebapiAsync work with redis connection

Author: joanhe

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

@@ -9,12 +9,27 @@
 namespace Magento\WebapiAsync\Code\Generator\Config\RemoteServiceReader;
 
 use Magento\AsynchronousOperations\Model\ConfigInterface as WebApiAsyncConfig;
+use Magento\Framework\MessageQueue\DefaultValueProvider;
 
 /**
  * Remote service reader with auto generated configuration for queue_consumer.xml
  */
 class Consumer implements \Magento\Framework\Config\ReaderInterface
 {
+    /**
+     * @var DefaultValueProvider
+     */
+    private $defaultValueProvider;
+
+    /**
+     * @param DefaultValueProvider $defaultValueProvider
+     */
+    public function __construct(
+        DefaultValueProvider $defaultValueProvider
+    ) {
+        $this->defaultValueProvider = $defaultValueProvider;
+    }
+
     /**
      * Generate consumer configuration based on remote services declarations
      *
@@ -31,7 +46,7 @@ public function read($scope = null)
                 'name'             => $topicName,
                 'queue'            => $topicName,
                 'consumerInstance' => WebApiAsyncConfig::DEFAULT_CONSUMER_INSTANCE,
-                'connection'       => WebApiAsyncConfig::DEFAULT_CONSUMER_CONNECTION,
+                'connection'       => $this->getConnection(),
                 'maxMessages'      => WebApiAsyncConfig::DEFAULT_CONSUMER_MAX_MESSAGE,
                 'handlers'         => [],
                 'maxIdleTime'      => null,
@@ -41,4 +56,16 @@ 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;
+    }
 }

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

@@ -10,6 +10,7 @@
 
 use Magento\Framework\MessageQueue\PublisherInterface;
 use Magento\Framework\OsInfo;
+use Magento\TestFramework\Helper\Bootstrap;
 use Magento\TestFramework\Helper\Amqp;
 
 /**
@@ -52,6 +53,11 @@ class PublisherConsumerController
      */
     private $amqpHelper;
 
+    /**
+     * @var ClearQueueProcessor
+     */
+    private $clearQueueProcessor;
+
     /**
      * PublisherConsumerController constructor.
      * @param PublisherInterface $publisher
@@ -61,6 +67,7 @@ class PublisherConsumerController
      * @param array $consumers
      * @param array $appInitParams
      * @param null|int $maxMessages
+     * @param ClearQueueProcessor $clearQueueProcessor
      */
     public function __construct(
         PublisherInterface $publisher,
@@ -69,7 +76,8 @@ public function __construct(
         $logFilePath,
         $consumers,
         $appInitParams,
-        $maxMessages = null
+        $maxMessages = null,
+        ClearQueueProcessor $clearQueueProcessor = null
     ) {
         $this->consumers = $consumers;
         $this->publisher = $publisher;
@@ -78,6 +86,9 @@ public function __construct(
         $this->osInfo = $osInfo;
         $this->appInitParams = $appInitParams;
         $this->amqpHelper = $amqpHelper;
+        $this->clearQueueProcessor = $clearQueueProcessor
+            ?: Bootstrap::getObjectManager()->get(ClearQueueProcessor::class);
+
     }
 
     /**
@@ -90,12 +101,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 +129,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.'
-            );
-        }
     }
 
     /**