update

workbunny · Aug 19, 2024 · 08cb7fe · 08cb7fe
1 parent 14214c3
commit 08cb7fe
Show file tree

Hide file tree

Showing 4 changed files with 52 additions and 12 deletions.
diff --git a/src/Builders/AbstractBuilder.php b/src/Builders/AbstractBuilder.php
@@ -16,6 +16,11 @@ abstract class AbstractBuilder
      */
     private static bool $reuseConnection = false;
 
+    /**
+     * @var bool
+     */
+    private static bool $reuseChannel = false;
+
     /**
      * builder对象池
      *
@@ -47,6 +52,7 @@ public function __construct()
         $this->setBuilderName(get_called_class());
         $config = config('plugin.workbunny.webman-rabbitmq.app');
         self::$reuseConnection = $config['reuse_connection'] ?? false;
+        self::$reuseChannel = $config['reuse_channel'] ?? false;
         $this->setConnection(new Connection($config));
         $this->setBuilderConfig(new BuilderConfig());
     }
@@ -61,6 +67,16 @@ public static function isReuseConnection(): bool
         return self::$reuseConnection;
     }
 
+    /**
+     * 是否复用channel
+     *
+     * @return bool
+     */
+    public static function isReuseChannel(): bool
+    {
+        return self::$reuseChannel;
+    }
+
     /**
      * builder单例
      *

diff --git a/src/Clients/Traits/ClientMethods.php b/src/Clients/Traits/ClientMethods.php
@@ -42,13 +42,14 @@ public function getChannels(): array
     /**
      * 获取一个可用的通道
      *
+     * @param bool $reuse
      * @return CurrentChannel|PromiseInterface
      */
-    public function catchChannel(): CurrentChannel|PromiseInterface
+    public function catchChannel(bool $reuse = false): CurrentChannel|PromiseInterface
     {
         $resChannel = null;
         // 从已创建的频道中获取一个可用的频道
-        $channels = $this->getChannels();
+        $channels = $reuse ? $this->getChannels() : [];
         foreach ($channels as $channel) {
             if (
                 $channel instanceof CurrentChannel and

diff --git a/src/Connection.php b/src/Connection.php
@@ -11,6 +11,7 @@
 use React\Promise\Deferred;
 use React\Promise\PromiseInterface;
 use Throwable;
+use Workbunny\WebmanRabbitMQ\Builders\AbstractBuilder;
 use Workbunny\WebmanRabbitMQ\Clients\AsyncClient;
 use Workbunny\WebmanRabbitMQ\Clients\SyncClient;
 use Workbunny\WebmanRabbitMQ\Exceptions\WebmanRabbitMQAsyncPublishException;
@@ -63,7 +64,7 @@ public function getSyncClient(): SyncClient
     public function getErrorCallback(): ?callable
     {
         $errorCallback = $this->_config['error_callback'] ?? null;
-        if(!is_callable($errorCallback) and !is_null($errorCallback)){
+        if (!is_callable($errorCallback) and !is_null($errorCallback)) {
             $errorCallback = null;
         }
         return $errorCallback;
@@ -94,16 +95,36 @@ public function disconnect(?AbstractClient $client, ?Throwable $throwable = null
         $replyCode = $throwable instanceof ClientException ? $throwable->getCode() : 0;
         $replyText = $throwable instanceof ClientException ? $throwable->getMessage() : '';
         try {
-            switch ($client) {
-                case $this->getAsyncClient():
-                    $this->getAsyncClient()->syncDisconnect($replyCode, $replyText);
+            switch (true) {
+                case $client instanceof AsyncClient:
+                    foreach ($client->getChannels() as $channelId => $channel) {
+                        if ($client->isConnected()) {
+                            $client->syncChannelClose($channelId, $replyCode, $replyText, 0, 0);
+                        }
+                        $client->removeChannel($channelId);
+                    }
+                    if ($client->isConnected()) {
+                        $client->syncDisconnect($replyCode, $replyText);
+                    }
                     break;
-                case $this->getSyncClient():
-                    $this->getSyncClient()->disconnect($replyCode, $replyText);
+                case $client instanceof SyncClient:
+                    foreach ($client->getChannels() as $channelId => $channel) {
+                        if ($client->isConnected()) {
+                            $channel->close($replyCode, $replyText);
+                        }
+                        $client->removeChannel($channelId);
+                    }
+                    if ($client->isConnected()) {
+                        $client->disconnect($replyCode, $replyText);
+                    }
                     break;
-                case null:
-                    $this->disconnect($this->getAsyncClient(), $throwable);
-                    $this->disconnect($this->getSyncClient(), $throwable);
+                case $client === null:
+                    if ($this->getAsyncClient()) {
+                        $this->disconnect($this->getAsyncClient(), $throwable);
+                    }
+                    if ($this->getSyncClient()) {
+                        $this->disconnect($this->getSyncClient(), $throwable);
+                    }
                     break;
                 default:
                     return;
@@ -252,7 +273,7 @@ public function syncPublish(BuilderConfig $config, bool $close = false): bool
     {
         try {
             if ($this->getSyncClient()->isConnected()) {
-                $channel = $this->getSyncClient()->catchChannel();
+                $channel = $this->getSyncClient()->catchChannel(AbstractBuilder::isReuseChannel());
             } else {
                 try {
                     $channel = $this->getSyncClient()->connect()->catchChannel();

diff --git a/src/config/plugin/workbunny/webman-rabbitmq/app.php b/src/config/plugin/workbunny/webman-rabbitmq/app.php
@@ -21,6 +21,8 @@
     },
     // 复用连接
     'reuse_connection'   => false,
+    // 复用通道
+    'reuse_channel'      => true,
     // AMQPS 如需使用AMQPS请取消注释
 //    'ssl'                => [
 //        'cafile'      => 'ca.pem',