Merge pull request #43 from utopia-php/feat-pool-adapter

Feat pool adapter

Author: abnegate

.github/workflows/tests.yml

@@ -16,6 +16,7 @@ jobs:
         adapter:
           [
             AMQP,
+            Pool,
             SwooleRedisCluster,
             Swoole,
             Workerman,

Dockerfile

@@ -1,4 +1,4 @@
-FROM composer as composer
+FROM composer AS composer
 
 WORKDIR /usr/local/src/
 

composer.json

@@ -29,6 +29,7 @@
         "utopia-php/cli": "0.15.*",
         "utopia-php/framework": "0.33.*",
         "utopia-php/telemetry": "0.1.*",
+        "utopia-php/pools": "0.8.*",
         "utopia-php/fetch": "0.4.*"
     },
     "require-dev": {

composer.lock

@@ -0,0 +1,56 @@
+<?php
+
+namespace Utopia\Queue\Broker;
+
+use Utopia\Queue\Consumer;
+use Utopia\Queue\Publisher;
+use Utopia\Queue\Queue;
+use Utopia\Pools\Pool as UtopiaPool;
+
+readonly class Pool implements Publisher, Consumer
+{
+    public function __construct(
+        private ?UtopiaPool $publisher = null,
+        private ?UtopiaPool $consumer = null,
+    ) {
+    }
+
+    public function enqueue(Queue $queue, array $payload): bool
+    {
+        return $this->delegatePublish(__FUNCTION__, \func_get_args());
+    }
+
+    public function retry(Queue $queue, ?int $limit = null): void
+    {
+        $this->delegatePublish(__FUNCTION__, \func_get_args());
+    }
+
+    public function getQueueSize(Queue $queue, bool $failedJobs = false): int
+    {
+        return $this->delegatePublish(__FUNCTION__, \func_get_args());
+    }
+
+    public function consume(Queue $queue, callable $messageCallback, callable $successCallback, callable $errorCallback): void
+    {
+        $this->delegateConsumer(__FUNCTION__, \func_get_args());
+    }
+
+    public function close(): void
+    {
+        $this->delegateConsumer(__FUNCTION__, \func_get_args());
+    }
+
+    protected function delegatePublish(string $method, array $args): mixed
+    {
+        return $this->publisher?->use(function (Publisher $adapter) use ($method, $args) {
+            return $adapter->$method(...$args);
+        });
+    }
+
+    protected function delegateConsumer(string $method, array $args): mixed
+    {
+        return $this->consumer?->use(function (Consumer $adapter) use ($method, $args) {
+            return $adapter->$method(...$args);
+        });
+    }
+}

src/Queue/Broker/Pool.php

@@ -1,8 +1,7 @@
 <?php
 
-namespace Queue\E2E\Adapter;
+namespace Tests\E2E\Adapter;
 
-use Tests\E2E\Adapter\Base;
 use Utopia\Queue\Broker\AMQP;
 use Utopia\Queue\Publisher;
 use Utopia\Queue\Queue;

tests/Queue/E2E/Adapter/AMQPTest.php

@@ -0,0 +1,27 @@
+<?php
+
+namespace Tests\E2E\Adapter;
+
+use Utopia\Pools\Pool as UtopiaPool;
+use Utopia\Queue\Broker\Pool;
+use Utopia\Queue\Broker\Redis as RedisBroker;
+use Utopia\Queue\Connection\Redis;
+use Utopia\Queue\Publisher;
+use Utopia\Queue\Queue;
+
+class PoolTest extends Base
+{
+    protected function getPublisher(): Publisher
+    {
+        $pool = new UtopiaPool('redis', 1, function () {
+            return new RedisBroker(new Redis('redis', 6379));
+        });
+
+        return new Pool($pool, $pool);
+    }
+
+    protected function getQueue(): Queue
+    {
+        return new Queue('pool');
+    }
+}

tests/Queue/E2E/Adapter/PoolTest.php

@@ -1,8 +1,7 @@
 <?php
 
-namespace Queue\E2E\Adapter;
+namespace Tests\E2E\Adapter;
 
-use Tests\E2E\Adapter\Base;
 use Utopia\Queue\Broker\Redis;
 use Utopia\Queue\Connection\RedisCluster;
 use Utopia\Queue\Publisher;