AC-14558::Migration form RabbitMQ to Apache ActiveMQ

Author: Dnyaneshwar S Jambhulkar

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

@@ -85,16 +85,12 @@ public function process($maxNumberOfMessages = null)
         if (!isset($maxNumberOfMessages)) {
             $queue->subscribe($this->getTransactionCallback($queue));
         } else {
-            $connectionName = $this->consumerConfig
-                              ->getConsumer($this->configuration->getConsumerName())
-                              ->getConnection();
             $this->invoker->invoke(
                 $queue,
                 $maxNumberOfMessages,
                 $this->getTransactionCallback($queue),
                 $maxIdleTime,
-                $sleep,
-                $connectionName
+                $sleep
             );
         }
 

app/code/Magento/Stomp/Model/Rpc/Publisher.php

@@ -111,7 +111,7 @@ public function publish($topicName, $data)
         $publisher = $this->publisherConfig->getPublisher($topicName);
         $connectionName = $publisher->getConnection()->getName();
         $queue = $this->queueRepository->get($connectionName, $replyTo);
-        $responseMessage = $queue->push($envelope);
+        $responseMessage = $queue->callRpc($envelope);
         return $this->messageEncoder->decode($topicName, $responseMessage, false);
     }
 }

lib/internal/Magento/Framework/MessageQueue/CallbackInvoker.php

@@ -6,6 +6,7 @@
 
 namespace Magento\Framework\MessageQueue;
 
+use Magento\Framework\App\ObjectManager;
 use Magento\Framework\MessageQueue\PoisonPill\PoisonPillCompareInterface;
 use Magento\Framework\MessageQueue\PoisonPill\PoisonPillReadInterface;
 use Magento\Framework\App\DeploymentConfig;
@@ -35,19 +36,28 @@ class CallbackInvoker implements CallbackInvokerInterface
      */
     private $deploymentConfig;
 
+    /**
+     * @var DefaultValueProvider
+     */
+    private $defaultValueProvider;
+
     /**
      * @param PoisonPillReadInterface $poisonPillRead
      * @param PoisonPillCompareInterface $poisonPillCompare
      * @param DeploymentConfig $deploymentConfig
+     * @param DefaultValueProvider|null $defaultValueProvider
      */
     public function __construct(
         PoisonPillReadInterface $poisonPillRead,
         PoisonPillCompareInterface $poisonPillCompare,
-        DeploymentConfig $deploymentConfig
+        DeploymentConfig $deploymentConfig,
+        DefaultValueProvider $defaultValueProvider
     ) {
         $this->poisonPillRead = $poisonPillRead;
         $this->poisonPillCompare = $poisonPillCompare;
         $this->deploymentConfig = $deploymentConfig;
+        $this->defaultValueProvider = $defaultValueProvider
+            ?? ObjectManager::getInstance()->get(DefaultValueProvider::class);
     }
 
     /**
@@ -69,12 +79,12 @@ public function invoke(
         $maxNumberOfMessages,
         $callback,
         $maxIdleTime = null,
-        $sleep = null,
-        $connectionName = 'amqp'
+        $sleep = null
     ) {
         $this->poisonPillVersion = $this->poisonPillRead->getLatestVersion();
         $sleep = (int) $sleep ?: 1;
         $maxIdleTime = $maxIdleTime ? (int) $maxIdleTime : PHP_INT_MAX;
+        $connectionName = $this->defaultValueProvider->getConnection();
         if ($connectionName === 'stomp') {
             $queue->subscribeQueue();
         }

lib/internal/Magento/Framework/MessageQueue/CallbackInvokerInterface.php

@@ -29,7 +29,6 @@ public function invoke(
         $maxNumberOfMessages,
         $callback,
         $maxIdleTime = null,
-        $sleep = null,
-        $connectionName = 'amqp'
+        $sleep = null
     );
 }

lib/internal/Magento/Framework/MessageQueue/Consumer.php

@@ -134,16 +134,12 @@ public function process($maxNumberOfMessages = null)
         if (!isset($maxNumberOfMessages)) {
             $queue->subscribe($this->getTransactionCallback($queue));
         } else {
-            $connectionName = $this->consumerConfig
-                ->getConsumer($this->configuration->getConsumerName())
-                ->getConnection();
             $this->invoker->invoke(
                 $queue,
                 $maxNumberOfMessages,
                 $this->getTransactionCallback($queue),
                 $maxIdleTime,
-                $sleep,
-                $connectionName
+                $sleep
             );
         }
     }
@@ -214,7 +210,7 @@ private function sendResponse(EnvelopeInterface $envelope)
         $messageProperties = $envelope->getProperties();
         $connectionName = $this->consumerConfig->getConsumer($this->configuration->getConsumerName())->getConnection();
         $queue = $this->queueRepository->get($connectionName, $messageProperties['reply_to']);
-        $queue->push($envelope);
+        $queue->callRpc($envelope);
     }
 
     /**

lib/internal/Magento/Framework/MessageQueue/QueueInterface.php

@@ -55,7 +55,7 @@ public function reject(EnvelopeInterface $envelope, $requeue = true, $rejectionM
      * Push message to queue directly, without using exchange
      *
      * @param EnvelopeInterface $envelope
-     * @return mixed
+     * @return void
      * @since 103.0.0
      */
     public function push(EnvelopeInterface $envelope);

lib/internal/Magento/Framework/Stomp/Bulk/Queue.php

@@ -87,7 +87,7 @@ public function push(BaseQueueInterface $queue, string $topic, array $envelopes)
         if ($isSync) {
             $responses = [];
             foreach ($envelopes as $envelope) {
-                $responses[] = $queue->push($envelope);
+                $responses[] = $queue->callRpc($envelope);
             }
             return $responses;
         }

lib/internal/Magento/Framework/Stomp/Queue.php

@@ -205,6 +205,26 @@ public function reject(EnvelopeInterface $envelope, $requeue = true, $rejectionM
      * @throws ConnectionLostException
      */
     public function push(EnvelopeInterface $envelope)
+    {
+        $stompClient = $this->getStompProducerClient();
+        $message = new Message($envelope->getBody(), $envelope->getProperties());
+        try{
+            $stompClient->send($this->queueName, $message);
+            $stompClient->disconnect();
+        }catch (\Stomp\Exception\StompException $e){
+            $this->logger->info("Stomp message push failed: '{$this->queueName}' error: {$e->getMessage()}");
+        }
+    }
+
+    /**
+     * Push message to queue and read message from queue
+     *
+     * @param EnvelopeInterface $envelope
+     * @return string|null
+     * @throws ConnectionLostException
+     * @throws LocalizedException
+     */
+    public function callRpc(EnvelopeInterface $envelope)
     {
         $stompClient = $this->getStompProducerClient();
         $properties = $envelope->getProperties();
@@ -224,7 +244,7 @@ public function push(EnvelopeInterface $envelope)
                 return $message->getBody();
             }
         }catch (\Stomp\Exception\StompException $e){
-            return null;
+            $this->logger->info("Stomp rpc message push failed: '{$this->queueName}' error: '{$e->getMessage()}'");
         }
 
         return null;