Added params connection_name which used to define amqp connection n…

…ame.
hyperf · Jan 8, 2025 · db5e248 · db5e248
1 parent 45f1c42
commit db5e248
Show file tree

Hide file tree

Showing 4 changed files with 31 additions and 3 deletions.
diff --git a/publish/amqp.php b/publish/amqp.php
@@ -44,6 +44,7 @@
             'channel_rpc_timeout' => 0.0,
             'close_on_destruct' => false,
             'max_idle_channels' => 10,
+            'connection_name' => null,
         ],
     ],
 ];
diff --git a/src/AMQPConnection.php b/src/AMQPConnection.php
@@ -24,6 +24,7 @@
 use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Channel\Frame;
 use PhpAmqpLib\Connection\AbstractConnection;
+use PhpAmqpLib\Connection\AMQPConnectionConfig;
 use PhpAmqpLib\Exception\AMQPRuntimeException;
 use PhpAmqpLib\Exception\AMQPTimeoutException;
 use PhpAmqpLib\Wire\AMQPWriter;
@@ -68,13 +69,14 @@ public function __construct(
         ?AbstractIO $io = null,
         int $heartbeat = 0,
         int $connection_timeout = 0,
-        float $channel_rpc_timeout = 0.0
+        float $channel_rpc_timeout = 0.0,
+        ?AMQPConnectionConfig $config = null
     ) {
         $this->channelManager = new ChannelManager(16);
         $this->channelManager->get(0, true);
         $this->chan = $this->channelManager->make(65535);
 
-        parent::__construct($user, $password, $vhost, $insist, $login_method, $login_response, $locale, $io, $heartbeat, $connection_timeout, $channel_rpc_timeout);
+        parent::__construct($user, $password, $vhost, $insist, $login_method, $login_response, $locale, $io, $heartbeat, $connection_timeout, $channel_rpc_timeout, $config);
 
         $this->pool = new Channel(static::CHANNEL_POOL_LENGTH);
         $this->confirmPool = new Channel(static::CONFIRM_CHANNEL_POOL_LENGTH);

diff --git a/src/ConnectionFactory.php b/src/ConnectionFactory.php
@@ -20,6 +20,7 @@
 use Hyperf\Contract\StdoutLoggerInterface;
 use Hyperf\Coroutine\Locker;
 use InvalidArgumentException;
+use PhpAmqpLib\Connection\AMQPConnectionConfig;
 use PhpAmqpLib\Wire\IO\AbstractIO;
 use Psr\Container\ContainerInterface;
 
@@ -91,6 +92,12 @@ public function make(array $config): AMQPConnection
         $params = new Params(Arr::get($config, 'params', []));
         $io = $this->makeIO($config, $params);
 
+        $amqpConfig = null;
+        if (! empty($params->getConnectionName())) {
+            $amqpConfig = new AMQPConnectionConfig();
+            $amqpConfig->setConnectionName($params->getConnectionName());
+        }
+
         $connection = new AMQPConnection(
             $user,
             $password,
@@ -102,7 +109,8 @@ public function make(array $config): AMQPConnection
             $io,
             $params->getHeartbeat(),
             $params->getConnectionTimeout(),
-            $params->getChannelRpcTimeout()
+            $params->getChannelRpcTimeout(),
+            $amqpConfig
         );
 
         return $connection->setParams($params)

diff --git a/src/Params.php b/src/Params.php
@@ -34,6 +34,8 @@ class Params
 
     protected int $maxIdleChannels = 10;
 
+    protected ?string $connectionName = null;
+
     public function __construct(array $data)
     {
         if (isset($data['insist'])) {
@@ -75,6 +77,10 @@ public function __construct(array $data)
         if (isset($data['max_idle_channels'])) {
             $this->setMaxIdleChannels((int) $data['max_idle_channels']);
         }
+
+        if (isset($data['connection_name'])) {
+            $this->setConnectionName($data['connection_name']);
+        }
     }
 
     public function isInsist(): bool
@@ -186,4 +192,15 @@ public function setMaxIdleChannels(int $maxIdleChannels): static
         $this->maxIdleChannels = $maxIdleChannels;
         return $this;
     }
+
+    public function getConnectionName(): ?string
+    {
+        return $this->connectionName;
+    }
+
+    public function setConnectionName(?string $connectionName): static
+    {
+        $this->connectionName = $connectionName;
+        return $this;
+    }
 }