Merge pull request #26 from utopia-php/worker-id-inject

Inject workerId as resource, record received timestamp

src/Queue/Connection/RedisCluster.php

@@ -6,7 +6,6 @@
 
 class RedisCluster implements Connection
 {
-
     protected array $seeds;
     protected ?\RedisCluster $redis = null;
 

src/Queue/Message.php

@@ -8,6 +8,7 @@ class Message
     protected string $queue;
     protected int $timestamp;
     protected array $payload;
+    protected float $receivedTimestamp;
 
     public function __construct(array $array = [])
     {
@@ -19,6 +20,7 @@ public function __construct(array $array = [])
         $this->queue = $array['queue'];
         $this->timestamp = $array['timestamp'];
         $this->payload = $array['payload'] ?? [];
+        $this->receivedTimestamp = (float)$array['receivedTimestamp'] ?? 0.;
     }
 
     public function setPid(string $pid): self
@@ -69,6 +71,17 @@ public function getPayload(): array
         return $this->payload;
     }
 
+    /**
+     * Timestamp recorded when this message was received by the worker.
+     * The delta between `$receivedTimestamp` and `$timestamp` is the amount of time the message was waiting in the queue.
+     *
+     * @return float
+     */
+    public function getReceivedTimestamp(): float
+    {
+        return $this->receivedTimestamp;
+    }
+
     public function asArray(): array
     {
         return [

src/Queue/Server.php

@@ -189,6 +189,7 @@ public function start(): self
         try {
             $this->adapter->workerStart(function (string $workerId) {
                 Console::success("[Worker] Worker {$workerId} is ready!");
+                self::setResource('workerId', fn () => $workerId);
                 if (!is_null($this->workerStartHook)) {
                     call_user_func_array($this->workerStartHook->getAction(), $this->getArguments($this->workerStartHook));
                 }
@@ -203,6 +204,7 @@ public function start(): self
                     }
 
                     $nextMessage['timestamp'] = (int)$nextMessage['timestamp'];
+                    $nextMessage['receivedTimestamp'] = microtime(true);
 
                     $message = new Message($nextMessage);
 

tests/Queue/E2E/Adapter/SwooleRedisClusterTest.php

@@ -4,7 +4,6 @@
 
 use Tests\E2E\Adapter\Base;
 use Utopia\Queue\Client;
-use Utopia\Queue\Connection\Redis;
 use Utopia\Queue\Connection\RedisCluster;
 
 class SwooleRedisClusterTest extends Base