feat: add concurrency manager

phpunit.xml

@@ -12,5 +12,8 @@
         <testsuite name="E2E">
             <directory>./tests/Queue/E2E/Adapter</directory>
         </testsuite>
+        <testsuite name="Unit">
+            <directory>./tests/Queue/Unit</directory>
+        </testsuite>
     </testsuites>
 </phpunit>

src/Queue/Adapter/Swoole/Redis.php

@@ -5,8 +5,9 @@
 use Swoole\Database\RedisConfig;
 use Swoole\Database\RedisPool;
 use Utopia\Queue\Connection\Redis as ConnectionRedis;
+use Utopia\Queue\Concurrency\Adapter;
 
-class Redis extends ConnectionRedis
+class Redis extends ConnectionRedis implements Adapter
 {
     protected RedisPool $pool;
 

src/Queue/Concurrency/Adapter.php

@@ -0,0 +1,11 @@
+<?php
+
+namespace Utopia\Queue\Concurrency;
+
+interface Adapter
+{
+    public function get(string $key): ?string;
+    public function set(string $key, string $value): void;
+    public function increment(string $key): int;
+    public function decrement(string $key): int;
+}

src/Queue/Concurrency/Manager.php

@@ -0,0 +1,37 @@
+<?php
+
+namespace Utopia\Queue\Concurrency;
+
+use Utopia\Queue\Message;
+
+abstract class Manager
+{
+    public function __construct(protected Adapter $adapter, protected int $limit = 1)
+    {
+    }
+
+    public function canProcessJob(Message $message): bool
+    {
+        $key = $this->getConcurrencyKey($message);
+        $value = $this->adapter->get($key);
+        if ($value === null) {
+            $this->adapter->set($key, "0");
+            $value = 0;
+        }
+        return \intval($value) < $this->limit;
+    }
+
+    public function startJob(Message $message): void
+    {
+        $key = $this->getConcurrencyKey($message);
+        $this->adapter->increment($key);
+    }
+
+    public function finishJob(Message $message): void
+    {
+        $key = $this->getConcurrencyKey($message);
+        $this->adapter->decrement($key);
+    }
+
+    abstract public function getConcurrencyKey(Message $message): string;
+}

src/Queue/Worker.php

@@ -8,6 +8,7 @@
 use Utopia\DI\Container;
 use Utopia\DI\Dependency;
 use Utopia\Servers\Base;
+use Utopia\Queue\Concurrency\Manager;
 
 class Worker extends Base
 {
@@ -146,6 +147,20 @@ public function start(): self
         return $this;
     }
 
+    protected ?Manager $concurrencyManager = null;
+
+    /**
+     * Set the concurrency manager
+     *
+     * @param Manager $manager
+     * @return self
+     */
+    public function setConcurrencyManager(Manager $manager): self
+    {
+        $this->concurrencyManager = $manager;
+        return $this;
+    }
+
     protected function lifecycle(Job $job, Message $message, array $nextMessage, Container $context, Connection $connection): static
     {
         Console::info("[Job] Received Job ({$message->getPid()}).");
@@ -154,21 +169,15 @@ protected function lifecycle(Job $job, Message $message, array $nextMessage, Con
 
         $connection->getConnection();
 
-        /**
-         * Move Job to Jobs and it's PID to the processing list.
-         */
-        $connection->setArray("{$this->adapter->namespace}.jobs.{$this->adapter->queue}.{$message->getPid()}", $nextMessage);
-        $connection->leftPush("{$this->adapter->namespace}.processing.{$this->adapter->queue}", $message->getPid());
-
-        /**
-         * Increment Total Jobs Received from Stats.
-         */
-        $connection->increment("{$this->adapter->namespace}.stats.{$this->adapter->queue}.total");
-
-        /**
-         * Increment Processing Jobs from Stats.
-         */
-        $connection->increment("{$this->adapter->namespace}.stats.{$this->adapter->queue}.processing");
+        // if ($this->concurrencyManager) {
+        //     if (!$this->concurrencyManager->canProcessJob($message)) {
+        //         // If we can't process the job due to concurrency limits, put it back in the queue
+        //         $connection->leftPush("{$this->adapter->namespace}.queue.{$this->adapter->queue}", json_encode($nextMessage));
+        //         Console::info("[Job] ({$message->getPid()}) postponed due to concurrency limits.");
+        //         return $this;
+        //     }
+        //     $this->concurrencyManager->startJob($message);
+        // }
 
         try {
             foreach (self::$init as $hook) { // Global init hooks
@@ -257,6 +266,10 @@ protected function lifecycle(Job $job, Message $message, array $nextMessage, Con
                 }
             }
         } finally {
+            // if ($this->concurrencyManager) {
+            //     $this->concurrencyManager->finishJob($message);
+            // }
+
             /**
              * Remove Job from Processing.
              */