AC-14558::Migration form RabbitMQ to Apache ActiveMQ

Author: Dnyaneshwar S Jambhulkar

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

@@ -210,7 +210,11 @@ 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->callRpc($envelope);
+        if($connectionName === 'stomp') {
+            $queue->callRpc($envelope);
+        } else {
+            $queue->push($envelope);
+        }
     }
 
     /**

lib/internal/Magento/Framework/MessageQueue/Test/Unit/ConsumerTest.php

@@ -26,6 +26,7 @@
 use PHPUnit\Framework\MockObject\MockObject;
 use PHPUnit\Framework\TestCase;
 use Psr\Log\LoggerInterface;
+use Magento\Framework\MessageQueue\DefaultValueProvider;
 
 /**
  * Unit test for Consumer class.
@@ -99,6 +100,11 @@ class ConsumerTest extends TestCase
      */
     private $deploymentConfig;
 
+    /**
+     * @var DefaultValueProvider|MockObject
+     */
+    private $defaultValueProvider;
+
     /**
      * Set up.
      *
@@ -123,6 +129,7 @@ protected function setUp(): void
             ->disableOriginalConstructor()
             ->getMockForAbstractClass();
         $this->deploymentConfig = $this->createMock(DeploymentConfig::class);
+        $this->defaultValueProvider = $this->createMock(DefaultValueProvider::class);
 
         $objectManager = new ObjectManager($this);
         $this->poisonPillCompare = $this->getMockBuilder(PoisonPillCompareInterface::class)
@@ -135,7 +142,8 @@ protected function setUp(): void
         $this->callbackInvoker = new CallbackInvoker(
             $this->poisonPillRead,
             $this->poisonPillCompare,
-            $this->deploymentConfig
+            $this->deploymentConfig,
+            $this->defaultValueProvider
         );
         $this->consumer = $objectManager->getObject(
             Consumer::class,
@@ -182,12 +190,15 @@ protected function setUp(): void
     public function testProcessWithNotFoundException()
     {
         $properties = ['topic_name' => 'topic.name'];
-        $topicConfig = [];
+        $topicConfig = ['is_synchronous' => true];
         $numberOfMessages = 1;
         $consumerName = 'consumer.name';
         $exceptionPhrase = new Phrase('Exception successfully thrown');
         $this->poisonPillRead->expects($this->atLeastOnce())->method('getLatestVersion')->willReturn('version-1');
         $this->poisonPillCompare->expects($this->atLeastOnce())->method('isLatestVersion')->willReturn(true);
+        $this->defaultValueProvider->expects($this->once())->method('getConnection')->willReturn('db');
+        $this->deploymentConfig->expects($this->any())->method('get')
+            ->with('queue/consumers_wait_for_messages', 1)->willReturn(1);
         $queue = $this->getMockBuilder(QueueInterface::class)
             ->disableOriginalConstructor()
             ->getMockForAbstractClass();
@@ -200,14 +211,6 @@ public function testProcessWithNotFoundException()
         $this->communicationConfig->expects($this->once())->method('getTopic')->with($properties['topic_name'])
             ->willReturn($topicConfig);
         $this->configuration->expects($this->atLeastOnce())->method('getConsumerName')->willReturn($consumerName);
-        $consumerConfigItem = $this->getMockBuilder(
-            \Magento\Framework\MessageQueue\Consumer\Config\ConsumerConfigItemInterface::class
-        )
-            ->disableOriginalConstructor()
-            ->getMockForAbstractClass();
-        $consumerConfigItem->expects($this->once())->method('getConnection')->willReturn('connection_name');
-        $this->consumerConfig->expects($this->once())->method('getConsumer')->with($consumerName)
-            ->willReturn($consumerConfigItem);
         $this->messageController->expects($this->once())->method('lock')->with($envelope, $consumerName)
             ->willThrowException(
                 new NotFoundException(
@@ -228,23 +231,17 @@ public function testProcessWithNotFoundException()
     public function testProcessWithGetMaxIdleTimeAndGetSleepConsumerConfigurations()
     {
         $numberOfMessages = 1;
-        $this->poisonPillRead->expects($this->atLeastOnce())->method('getLatestVersion');
+        $this->poisonPillRead->expects($this->atLeastOnce())->method('getLatestVersion')->willReturn('version-1');
+        $this->poisonPillCompare->expects($this->any())->method('isLatestVersion')->willReturn(true);
+        $this->defaultValueProvider->expects($this->once())->method('getConnection')->willReturn('db');
+        $this->deploymentConfig->expects($this->any())->method('get')
+            ->with('queue/consumers_wait_for_messages', 1)->willReturn(1);
         $queue = $this->getMockBuilder(\Magento\Framework\MessageQueue\QueueInterface::class)
             ->disableOriginalConstructor()->getMock();
         $this->configuration->expects($this->once())->method('getQueue')->willReturn($queue);
         $queue->expects($this->atMost(2))->method('dequeue')->willReturn(null);
         $this->configuration->expects($this->once())->method('getMaxIdleTime')->willReturn('2');
         $this->configuration->expects($this->once())->method('getSleep')->willReturn('2');
-        $consumerName = 'consumer.name';
-        $this->configuration->expects($this->once())->method('getConsumerName')->willReturn($consumerName);
-        $consumerConfigItem = $this->getMockBuilder(
-            \Magento\Framework\MessageQueue\Consumer\Config\ConsumerConfigItemInterface::class
-        )
-            ->disableOriginalConstructor()
-            ->getMockForAbstractClass();
-        $consumerConfigItem->expects($this->once())->method('getConnection')->willReturn('connection_name');
-        $this->consumerConfig->expects($this->once())->method('getConsumer')->with($consumerName)
-            ->willReturn($consumerConfigItem);
         $this->consumer->process($numberOfMessages);
     }
 }