Add Locking connection decorator for coroutine-safe access

composer.json

@@ -29,6 +29,7 @@
         "php": ">=8.3",
         "php-amqplib/php-amqplib": "^3.7",
         "utopia-php/di": "0.3.*",
+        "utopia-php/lock": "0.2.*",
         "utopia-php/servers": "0.4.*",
         "utopia-php/pools": "1.*",
         "utopia-php/telemetry": "0.4.*",

src/Queue/Connection/Locking.php

@@ -0,0 +1,152 @@
+<?php
+
+namespace Utopia\Queue\Connection;
+
+use Utopia\Lock\Lock;
+use Utopia\Lock\Mutex;
+use Utopia\Queue\Connection;
+
+/**
+ * Wraps any {@see Connection} and serializes every command behind a single
+ * lock, so that concurrent coroutines sharing one connection cannot interleave
+ * their requests and responses on the same socket.
+ *
+ * Outside of a coroutine there is no preemption, so the lock degrades to a
+ * plain in-process flag (see {@see Mutex}).
+ */
+class Locking implements Connection
+{
+    /**
+     * Wait forever when acquiring the lock; a command should never be dropped
+     * just because the connection is momentarily busy.
+     */
+    private const float ACQUIRE_TIMEOUT = -1;
+
+    public function __construct(
+        protected readonly Connection $connection,
+        protected readonly Lock $lock = new Mutex(),
+    ) {
+    }
+
+    /**
+     * Run a command while holding the lock, ensuring only one runs at a time.
+     *
+     * @template T
+     * @param callable(): T $command
+     * @return T
+     */
+    protected function synchronize(callable $command): mixed
+    {
+        return $this->lock->withLock($command, self::ACQUIRE_TIMEOUT);
+    }
+
+    public function rightPushArray(string $queue, array $payload): bool
+    {
+        return $this->synchronize(fn () => $this->connection->rightPushArray($queue, $payload));
+    }
+
+    public function rightPopArray(string $queue, int $timeout): array|false
+    {
+        return $this->synchronize(fn () => $this->connection->rightPopArray($queue, $timeout));
+    }
+
+    public function rightPopLeftPushArray(string $queue, string $destination, int $timeout): array|false
+    {
+        return $this->synchronize(fn () => $this->connection->rightPopLeftPushArray($queue, $destination, $timeout));
+    }
+
+    public function leftPushArray(string $queue, array $payload): bool
+    {
+        return $this->synchronize(fn () => $this->connection->leftPushArray($queue, $payload));
+    }
+
+    public function leftPopArray(string $queue, int $timeout): array|false
+    {
+        return $this->synchronize(fn () => $this->connection->leftPopArray($queue, $timeout));
+    }
+
+    public function rightPush(string $queue, string $payload): bool
+    {
+        return $this->synchronize(fn () => $this->connection->rightPush($queue, $payload));
+    }
+
+    public function rightPop(string $queue, int $timeout): string|false
+    {
+        return $this->synchronize(fn () => $this->connection->rightPop($queue, $timeout));
+    }
+
+    public function rightPopLeftPush(string $queue, string $destination, int $timeout): string|false
+    {
+        return $this->synchronize(fn () => $this->connection->rightPopLeftPush($queue, $destination, $timeout));
+    }
+
+    public function leftPush(string $queue, string $payload): bool
+    {
+        return $this->synchronize(fn () => $this->connection->leftPush($queue, $payload));
+    }
+
+    public function leftPop(string $queue, int $timeout): string|false
+    {
+        return $this->synchronize(fn () => $this->connection->leftPop($queue, $timeout));
+    }
+
+    public function listRemove(string $queue, string $key): bool
+    {
+        return $this->synchronize(fn () => $this->connection->listRemove($queue, $key));
+    }
+
+    public function listSize(string $key): int
+    {
+        return $this->synchronize(fn () => $this->connection->listSize($key));
+    }
+
+    public function listRange(string $key, int $total, int $offset): array
+    {
+        return $this->synchronize(fn () => $this->connection->listRange($key, $total, $offset));
+    }
+
+    public function remove(string $key): bool
+    {
+        return $this->synchronize(fn () => $this->connection->remove($key));
+    }
+
+    public function move(string $queue, string $destination): bool
+    {
+        return $this->synchronize(fn () => $this->connection->move($queue, $destination));
+    }
+
+    public function set(string $key, string $value, int $ttl = 0): bool
+    {
+        return $this->synchronize(fn () => $this->connection->set($key, $value, $ttl));
+    }
+
+    public function get(string $key): array|string|null
+    {
+        return $this->synchronize(fn () => $this->connection->get($key));
+    }
+
+    public function setArray(string $key, array $value, int $ttl = 0): bool
+    {
+        return $this->synchronize(fn () => $this->connection->setArray($key, $value, $ttl));
+    }
+
+    public function increment(string $key): int
+    {
+        return $this->synchronize(fn () => $this->connection->increment($key));
+    }
+
+    public function decrement(string $key): int
+    {
+        return $this->synchronize(fn () => $this->connection->decrement($key));
+    }
+
+    public function ping(): bool
+    {
+        return $this->synchronize(fn () => $this->connection->ping());
+    }
+
+    public function close(): void
+    {
+        $this->synchronize(fn () => $this->connection->close());
+    }
+}