AC-14558::Migration form RabbitMQ to Apache ActiveMQ

Dnyaneshwar S Jambhulkar · Dnyaneshwar S Jambhulkar · commit 2a5571abfa1a · 2025-08-21T15:59:54.000+05:30
diff --git a/app/code/Magento/Amqp/Setup/ConfigOptionsList.php b/app/code/Magento/Amqp/Setup/ConfigOptionsList.php
@@ -5,11 +5,11 @@
  */
 namespace Magento\Amqp\Setup;
 
+use Magento\Framework\App\DeploymentConfig;
 use Magento\Framework\Config\Data\ConfigData;
 use Magento\Framework\Config\File\ConfigFilePool;
 use Magento\Framework\Setup\ConfigOptionsListInterface;
 use Magento\Framework\Setup\Option\TextConfigOption;
-use Magento\Framework\App\DeploymentConfig;
 
 /**
  * Deployment configuration options needed for Setup application
diff --git a/app/code/Magento/AsynchronousOperations/Model/MassConsumer.php b/app/code/Magento/AsynchronousOperations/Model/MassConsumer.php
@@ -9,12 +9,12 @@
 
 use Magento\Framework\App\ObjectManager;
 use Magento\Framework\MessageQueue\CallbackInvokerInterface;
+use Magento\Framework\MessageQueue\Consumer\ConfigInterface as ConsumerConfig;
 use Magento\Framework\MessageQueue\ConsumerConfigurationInterface;
 use Magento\Framework\MessageQueue\ConsumerInterface;
 use Magento\Framework\MessageQueue\EnvelopeInterface;
 use Magento\Framework\MessageQueue\QueueInterface;
 use Magento\Framework\Registry;
-use Magento\Framework\MessageQueue\Consumer\ConfigInterface as ConsumerConfig;
 
 /**
  * Class Consumer used to process OperationInterface messages.
diff --git a/app/code/Magento/Stomp/Model/Bulk/Publisher.php b/app/code/Magento/Stomp/Model/Bulk/Publisher.php
@@ -8,13 +8,13 @@
 namespace Magento\Stomp\Model\Bulk;
 
 use Magento\Framework\MessageQueue\Bulk\Queue\QueueRepository as BulkQueueRepository;
-use Magento\Framework\MessageQueue\QueueRepository;
 use Magento\Framework\MessageQueue\EnvelopeFactory;
-use Magento\Framework\MessageQueue\MessageIdGeneratorInterface;
-use Magento\Framework\MessageQueue\PublisherInterface;
 use Magento\Framework\MessageQueue\MessageEncoder;
+use Magento\Framework\MessageQueue\MessageIdGeneratorInterface;
 use Magento\Framework\MessageQueue\MessageValidator;
 use Magento\Framework\MessageQueue\Publisher\ConfigInterface as PublisherConfig;
+use Magento\Framework\MessageQueue\PublisherInterface;
+use Magento\Framework\MessageQueue\QueueRepository;
 
 /**
  * A MessageQueue Publisher to handle publishing messages in bulk.
diff --git a/app/code/Magento/Stomp/Model/MassPublisher.php b/app/code/Magento/Stomp/Model/MassPublisher.php
@@ -7,14 +7,14 @@
 declare(strict_types=1);
 namespace Magento\Stomp\Model;
 
-use Magento\Framework\MessageQueue\MessageValidator;
+use Magento\AsynchronousOperations\Model\ConfigInterface as AsyncConfig;
+use Magento\Framework\MessageQueue\Bulk\Queue\QueueRepository as BulkQueueRepository;
+use Magento\Framework\MessageQueue\EnvelopeFactory;
 use Magento\Framework\MessageQueue\MessageEncoder;
+use Magento\Framework\MessageQueue\MessageIdGeneratorInterface;
+use Magento\Framework\MessageQueue\MessageValidator;
 use Magento\Framework\MessageQueue\Publisher\ConfigInterface as PublisherConfig;
-use Magento\Framework\MessageQueue\EnvelopeFactory;
-use Magento\AsynchronousOperations\Model\ConfigInterface as AsyncConfig;
 use Magento\Framework\MessageQueue\PublisherInterface;
-use Magento\Framework\MessageQueue\MessageIdGeneratorInterface;
-use Magento\Framework\MessageQueue\Bulk\Queue\QueueRepository as BulkQueueRepository;
 use Magento\Framework\MessageQueue\QueueRepository;
 
 /**
diff --git a/app/code/Magento/Stomp/Setup/ConfigOptionsList.php b/app/code/Magento/Stomp/Setup/ConfigOptionsList.php
@@ -7,11 +7,11 @@
 
 namespace Magento\Stomp\Setup;
 
+use Magento\Framework\App\DeploymentConfig;
 use Magento\Framework\Config\Data\ConfigData;
 use Magento\Framework\Config\File\ConfigFilePool;
 use Magento\Framework\Setup\ConfigOptionsListInterface;
 use Magento\Framework\Setup\Option\TextConfigOption;
-use Magento\Framework\App\DeploymentConfig;
 
 /**
  * Deployment configuration options needed for Setup application
diff --git a/app/code/Magento/Stomp/Setup/Recurring.php b/app/code/Magento/Stomp/Setup/Recurring.php
@@ -7,10 +7,10 @@
 
 namespace Magento\Stomp\Setup;
 
-use Magento\Framework\Stomp\TopologyInstaller;
 use Magento\Framework\Setup\InstallSchemaInterface;
 use Magento\Framework\Setup\ModuleContextInterface;
 use Magento\Framework\Setup\SchemaSetupInterface;
+use Magento\Framework\Stomp\TopologyInstaller;
 
 /**
  * Class Recurring used for installing queues in ActiveMq
diff --git a/app/code/Magento/Stomp/Test/Unit/Setup/ConfigOptionsListTest.php b/app/code/Magento/Stomp/Test/Unit/Setup/ConfigOptionsListTest.php
@@ -7,12 +7,12 @@
 
 namespace Magento\Stomp\Test\Unit\Setup;
 
-use Magento\Stomp\Setup\ConfigOptionsList;
-use Magento\Stomp\Setup\ConnectionValidator;
 use Magento\Framework\App\DeploymentConfig;
 use Magento\Framework\Config\Data\ConfigData;
 use Magento\Framework\Setup\Option\TextConfigOption;
 use Magento\Framework\TestFramework\Unit\Helper\ObjectManager;
+use Magento\Stomp\Setup\ConfigOptionsList;
+use Magento\Stomp\Setup\ConnectionValidator;
 use PHPUnit\Framework\MockObject\MockObject;
 use PHPUnit\Framework\TestCase;
 
diff --git a/dev/tests/integration/testsuite/Magento/Framework/MessageQueue/UseCase/DeprecatedFormat/AsyncMultiTopicsSeparateQueuesTest.php b/dev/tests/integration/testsuite/Magento/Framework/MessageQueue/UseCase/DeprecatedFormat/AsyncMultiTopicsSeparateQueuesTest.php
@@ -6,8 +6,8 @@
 namespace Magento\Framework\MessageQueue\UseCase\DeprecatedFormat;
 
 use Magento\Framework\MessageQueue\DefaultValueProvider;
-use Magento\TestFramework\Helper\Bootstrap;
 use Magento\Framework\MessageQueue\UseCase\QueueTestCaseAbstract;
+use Magento\TestFramework\Helper\Bootstrap;
 use Magento\TestModuleAsyncAmqp\Model\AsyncTestData;
 
 class AsyncMultiTopicsSeparateQueuesTest extends QueueTestCaseAbstract
diff --git a/dev/tests/integration/testsuite/Magento/Framework/Stomp/TopologyTest.php b/dev/tests/integration/testsuite/Magento/Framework/Stomp/TopologyTest.php
@@ -9,8 +9,8 @@
 namespace Magento\Framework\Stomp;
 
 use Magento\Framework\MessageQueue\DefaultValueProvider;
-use Magento\TestFramework\Helper\Stomp;
 use Magento\TestFramework\Helper\Bootstrap;
+use Magento\TestFramework\Helper\Stomp;
 use Magento\TestFramework\ObjectManager;
 use PHPUnit\Framework\TestCase;
 
diff --git a/lib/internal/Magento/Framework/Amqp/Config.php b/lib/internal/Magento/Framework/Amqp/Config.php
@@ -234,4 +234,14 @@ private function closeConnection()
             unset($this->connection);
         }
     }
+
+    /**
+     * Get connection name
+     *
+     * @return string
+     */
+    public function getConnectionName(): string
+    {
+        return $this->connectionName;
+    }
 }
diff --git a/lib/internal/Magento/Framework/Amqp/Queue.php b/lib/internal/Magento/Framework/Amqp/Queue.php
@@ -220,4 +220,14 @@ public function clearQueue(): int
     {
         throw new \BadMethodCallException('clearQueue is not supported in amqp queue.');
     }
+
+    /**
+     * Get connection name
+     *
+     * @return string
+     */
+    public function getConnectionName(): string
+    {
+        return $this->amqpConfig->getConnectionName();
+    }
 }
diff --git a/lib/internal/Magento/Framework/MessageQueue/BatchConsumer.php b/lib/internal/Magento/Framework/MessageQueue/BatchConsumer.php
@@ -6,8 +6,8 @@
 namespace Magento\Framework\MessageQueue;
 
 use Magento\Framework\App\ObjectManager;
-use Magento\Framework\MessageQueue\ConfigInterface as MessageQueueConfig;
 use Magento\Framework\App\ResourceConnection;
+use Magento\Framework\MessageQueue\ConfigInterface as MessageQueueConfig;
 use Magento\Framework\MessageQueue\Consumer\ConfigInterface as ConsumerConfig;
 
 /**
diff --git a/lib/internal/Magento/Framework/MessageQueue/CallbackInvoker.php b/lib/internal/Magento/Framework/MessageQueue/CallbackInvoker.php
@@ -6,10 +6,9 @@
 
 namespace Magento\Framework\MessageQueue;
 
-use Magento\Framework\App\ObjectManager;
+use Magento\Framework\App\DeploymentConfig;
 use Magento\Framework\MessageQueue\PoisonPill\PoisonPillCompareInterface;
 use Magento\Framework\MessageQueue\PoisonPill\PoisonPillReadInterface;
-use Magento\Framework\App\DeploymentConfig;
 
 /**
  * Class CallbackInvoker to invoke callbacks for consumer classes
@@ -36,28 +35,19 @@ 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,
-        DefaultValueProvider $defaultValueProvider
+        DeploymentConfig $deploymentConfig
     ) {
         $this->poisonPillRead = $poisonPillRead;
         $this->poisonPillCompare = $poisonPillCompare;
         $this->deploymentConfig = $deploymentConfig;
-        $this->defaultValueProvider = $defaultValueProvider
-            ?? ObjectManager::getInstance()->get(DefaultValueProvider::class);
     }
 
     /**
@@ -84,7 +74,7 @@ public function invoke(
         $this->poisonPillVersion = $this->poisonPillRead->getLatestVersion();
         $sleep = (int) $sleep ?: 1;
         $maxIdleTime = $maxIdleTime ? (int) $maxIdleTime : PHP_INT_MAX;
-        $connectionName = $this->defaultValueProvider->getConnection();
+        $connectionName = method_exists($queue, 'getConnectionName') ? $queue->getConnectionName(): null;
         if ($connectionName === 'stomp') {
             $queue->subscribeQueue();
         }
diff --git a/lib/internal/Magento/Framework/MessageQueue/Consumer.php b/lib/internal/Magento/Framework/MessageQueue/Consumer.php
@@ -210,11 +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']);
-        if($connectionName === 'stomp') {
-            $queue->callRpc($envelope);
-        } else {
-            $queue->push($envelope);
-        }
+        $queue->push($envelope);
     }
 
     /**
diff --git a/lib/internal/Magento/Framework/MessageQueue/Test/Unit/ConsumerTest.php b/lib/internal/Magento/Framework/MessageQueue/Test/Unit/ConsumerTest.php
@@ -14,6 +14,7 @@
 use Magento\Framework\MessageQueue\Consumer;
 use Magento\Framework\MessageQueue\Consumer\ConfigInterface;
 use Magento\Framework\MessageQueue\ConsumerConfigurationInterface;
+use Magento\Framework\MessageQueue\DefaultValueProvider;
 use Magento\Framework\MessageQueue\EnvelopeInterface;
 use Magento\Framework\MessageQueue\MessageController;
 use Magento\Framework\MessageQueue\MessageEncoder;
@@ -26,7 +27,6 @@
 use PHPUnit\Framework\MockObject\MockObject;
 use PHPUnit\Framework\TestCase;
 use Psr\Log\LoggerInterface;
-use Magento\Framework\MessageQueue\DefaultValueProvider;
 
 /**
  * Unit test for Consumer class.
diff --git a/lib/internal/Magento/Framework/Stomp/Bulk/Queue.php b/lib/internal/Magento/Framework/Stomp/Bulk/Queue.php
@@ -10,12 +10,12 @@
 use Magento\Framework\Communication\ConfigInterface as CommunicationConfigInterface;
 use Magento\Framework\Exception\LocalizedException;
 use Magento\Framework\MessageQueue\Bulk\Queue\QueueInterface;
+use Magento\Framework\MessageQueue\QueueInterface as BaseQueueInterface;
 use Magento\Framework\Stomp\Config;
-use Magento\Framework\Stomp\StompClientFactory;
 use Magento\Framework\Stomp\StompClient;
+use Magento\Framework\Stomp\StompClientFactory;
 use Stomp\Exception\StompException;
 use Stomp\Transport\Message;
-use Magento\Framework\MessageQueue\QueueInterface as BaseQueueInterface;
 
 /**
  * @api
@@ -125,4 +125,14 @@ private function getStompClient(): StompClient
         }
         return $this->stompClient;
     }
+
+    /**
+     * Get connection name
+     *
+     * @return string
+     */
+    public function getConnectionName(): string
+    {
+        return $this->stompConfig->getConnectionName();
+    }
 }
diff --git a/lib/internal/Magento/Framework/Stomp/Config.php b/lib/internal/Magento/Framework/Stomp/Config.php
@@ -7,11 +7,11 @@
 
 namespace Magento\Framework\Stomp;
 
-use Magento\Framework\Stomp\Connection\Factory as ConnectionFactory;
-use Magento\Framework\Stomp\Connection\FactoryOptions;
 use Magento\Framework\App\DeploymentConfig;
 use Magento\Framework\App\ObjectManager;
 use Magento\Framework\ObjectManager\ResetAfterRequestInterface;
+use Magento\Framework\Stomp\Connection\Factory as ConnectionFactory;
+use Magento\Framework\Stomp\Connection\FactoryOptions;
 use Stomp\Exception\ConnectionException;
 use Stomp\Network\Connection;
 
@@ -211,4 +211,14 @@ private function closeConnection(): void
             unset($this->connection);
         }
     }
+
+    /**
+     * Get connection name
+     *
+     * @return string
+     */
+    public function getConnectionName(): string
+    {
+        return$this->connectionName;
+    }
 }
diff --git a/lib/internal/Magento/Framework/Stomp/ConnectionTypeResolver.php b/lib/internal/Magento/Framework/Stomp/ConnectionTypeResolver.php
@@ -7,8 +7,8 @@
 
 namespace Magento\Framework\Stomp;
 
-use Magento\Framework\MessageQueue\ConnectionTypeResolverInterface;
 use Magento\Framework\App\DeploymentConfig;
+use Magento\Framework\MessageQueue\ConnectionTypeResolverInterface;
 
 /**
  * Stomp connection type resolver.
diff --git a/lib/internal/Magento/Framework/Stomp/Queue.php b/lib/internal/Magento/Framework/Stomp/Queue.php
@@ -12,13 +12,13 @@
 use Magento\Framework\Communication\ConfigInterface as CommunicationConfigInterface;
 use Magento\Framework\Exception\LocalizedException;
 use Magento\Framework\MessageQueue\ConnectionLostException;
+use Magento\Framework\MessageQueue\EnvelopeFactory;
 use Magento\Framework\MessageQueue\EnvelopeInterface;
 use Magento\Framework\MessageQueue\QueueInterface;
-use Magento\Framework\MessageQueue\EnvelopeFactory;
 use Magento\Framework\Phrase;
 use Psr\Log\LoggerInterface;
-use Stomp\Transport\Message;
 use Stomp\Transport\Frame;
+use Stomp\Transport\Message;
 
 /**
  * @api
@@ -367,4 +367,14 @@ private function getStompProducerClient(): StompClient
         }
         return $this->stompProducerClient;
     }
+
+    /**
+     * Get connection name
+     *
+     * @return string
+     */
+    public function getConnectionName(): string
+    {
+        return $this->stompConfig->getConnectionName();
+    }
 }
diff --git a/lib/internal/Magento/Framework/Stomp/StompClient.php b/lib/internal/Magento/Framework/Stomp/StompClient.php
@@ -7,13 +7,13 @@
 
 namespace Magento\Framework\Stomp;
 
+use Psr\Log\LoggerInterface;
 use Stomp\Client;
 use Stomp\Exception\StompException;
+use Stomp\Network\Observer\HeartbeatEmitter;
 use Stomp\StatefulStomp;
-use Stomp\Transport\Message;
 use Stomp\Transport\Frame;
-use Stomp\Network\Observer\HeartbeatEmitter;
-use Psr\Log\LoggerInterface;
+use Stomp\Transport\Message;
 
 /**
  * Wrapper StompClient class for stomp connection
diff --git a/lib/internal/Magento/Framework/Stomp/Test/Unit/Bulk/QueueTest.php b/lib/internal/Magento/Framework/Stomp/Test/Unit/Bulk/QueueTest.php
@@ -7,13 +7,13 @@
 
 namespace Magento\Framework\Stomp\Test\Unit\Bulk;
 
+use Magento\Framework\Communication\ConfigInterface as CommunicationConfigInterface;
+use Magento\Framework\MessageQueue\EnvelopeInterface;
+use Magento\Framework\MessageQueue\QueueInterface as BaseQueueInterface;
 use Magento\Framework\Stomp\Bulk\Queue;
 use Magento\Framework\Stomp\Config;
 use Magento\Framework\Stomp\StompClient;
 use Magento\Framework\Stomp\StompClientFactory;
-use Magento\Framework\MessageQueue\EnvelopeInterface;
-use Magento\Framework\MessageQueue\QueueInterface as BaseQueueInterface;
-use Magento\Framework\Communication\ConfigInterface as CommunicationConfigInterface;
 use PHPUnit\Framework\TestCase;
 use Stomp\Transport\Message;
 
diff --git a/lib/internal/Magento/Framework/Stomp/Test/Unit/ConfigPoolTest.php b/lib/internal/Magento/Framework/Stomp/Test/Unit/ConfigPoolTest.php
@@ -11,9 +11,9 @@
 use Magento\Framework\Stomp\ConfigFactory;
 use Magento\Framework\Stomp\ConfigPool;
 use PHPUnit\Framework\MockObject\Exception;
-use Stomp\Network\Connection;
 use PHPUnit\Framework\MockObject\MockObject;
 use PHPUnit\Framework\TestCase;
+use Stomp\Network\Connection;
 
 class ConfigPoolTest extends TestCase
 {
diff --git a/lib/internal/Magento/Framework/Stomp/Test/Unit/ConfigTest.php b/lib/internal/Magento/Framework/Stomp/Test/Unit/ConfigTest.php
diff --git a/lib/internal/Magento/Framework/Stomp/Test/Unit/ConnectionTypeResolverTest.php b/lib/internal/Magento/Framework/Stomp/Test/Unit/ConnectionTypeResolverTest.php
diff --git a/lib/internal/Magento/Framework/Stomp/Test/Unit/Topology/QueueInstallerTest.php b/lib/internal/Magento/Framework/Stomp/Test/Unit/Topology/QueueInstallerTest.php
diff --git a/lib/internal/Magento/Framework/Stomp/Test/Unit/TopologyInstallerTest.php b/lib/internal/Magento/Framework/Stomp/Test/Unit/TopologyInstallerTest.php
diff --git a/lib/internal/Magento/Framework/Stomp/Topology/QueueInstaller.php b/lib/internal/Magento/Framework/Stomp/Topology/QueueInstaller.php
diff --git a/lib/internal/Magento/Framework/Stomp/TopologyInstaller.php b/lib/internal/Magento/Framework/Stomp/TopologyInstaller.php

Original file line number	Diff line number	Diff line change
`@@ -234,4 +234,14 @@ private function closeConnection()`
`234`	`234`	`unset($this->connection);`
`235`	`235`	`}`
`236`	`236`	`}`
	`237`	`+`
	`238`	`+ /**`
	`239`	`+ * Get connection name`
	`240`	`+ *`
	`241`	`+ * @return string`
	`242`	`+ */`
	`243`	`+ public function getConnectionName(): string`
	`244`	`+ {`
	`245`	`+ return $this->connectionName;`
	`246`	`+ }`
`237`	`247`	`}`