Abstract swoole with channel

Author: abnegate

composer.json

@@ -33,14 +33,15 @@
         "utopia-php/fetch": "0.4.*"
     },
     "require-dev": {
-        "ext-redis": "*",
         "swoole/ide-helper": "5.1.7",
         "phpunit/phpunit": "9.*",
         "laravel/pint": "1.*",
         "workerman/workerman": "4.*",
         "phpstan/phpstan": "1.*"
     },
     "suggest": {
+        "ext-redis": "Needed to support Redis.",
+        "ext-amqp": "Needed to support AMQP.",
         "ext-swoole": "Needed to support Swoole.",
         "workerman/workerman": "Needed to support Workerman."
     },

composer.lock

@@ -58,6 +58,11 @@ public function __construct(
     ) {
     }
 
+    public function getConnectionType(): string
+    {
+        return AMQPStreamConnection::class;
+    }
+
     /**
      * Enable or disable waiting for publisher confirms.
      */
@@ -281,7 +286,7 @@ protected function withChannel(callable $callback): void
             : 0;
 
         $createChannel = function (): AMQPChannel {
-            $connection = new AMQPStreamConnection(
+            $connection = new ($this->getConnectionType())(
                 $this->host,
                 $this->port,
                 $this->user,

src/Queue/Broker/AMQP.php

@@ -2,56 +2,12 @@
 
 namespace Utopia\Queue\Broker;
 
-use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Connection\AMQPSwooleConnection;
 
 class AMQPSwoole extends AMQP
 {
-    /**
-     * Override the withChannel method to use AMQPSwooleConnection instead of AMQPStreamConnection
-     *
-     * @param callable(AMQPChannel $channel): void $callback
-     * @throws \Exception
-     */
-    protected function withChannel(callable $callback): void
+    public function getConnectionType(): string
     {
-        $createChannel = function (): AMQPChannel {
-            $connection = new AMQPSwooleConnection(
-                $this->host,
-                $this->port,
-                $this->user,
-                $this->password,
-                $this->vhost,
-                connection_timeout: $this->connectTimeout,
-                read_write_timeout: $this->readWriteTimeout,
-                heartbeat: $this->heartbeat,
-            );
-
-            if (\is_callable($this->connectionConfigHook)) {
-                ($this->connectionConfigHook)($connection);
-            }
-
-            $channel = $connection->channel();
-
-            if (\is_callable($this->channelConfigHook)) {
-                ($this->channelConfigHook)($channel);
-            }
-
-            return $channel;
-        };
-
-        if (!$this->channel) {
-            $this->channel = $createChannel();
-        }
-
-        try {
-            $callback($this->channel);
-        } catch (\Throwable) {
-            // createChannel() might throw, in that case set the channel to `null` first.
-            $this->channel = null;
-            // try creating a new connection once, if this still fails, throw the error
-            $this->channel = $createChannel();
-            $callback($this->channel);
-        }
+        return AMQPSwooleConnection::class;
     }
 }