feat(redis): expose reconnect success callback

Author: ChiragAgg5k

src/Queue/Broker/Redis.php

@@ -19,6 +19,10 @@ class Redis implements Publisher, Consumer
      * @var (callable(Queue, \Throwable, int, int): void)|null
      */
     private $reconnectCallback = null;
+    /**
+     * @var (callable(Queue, int): void)|null
+     */
+    private $reconnectSuccessCallback = null;
 
     public function __construct(private readonly Connection $connection)
     {
@@ -31,6 +35,13 @@ public function setReconnectCallback(?callable $callback): self
         return $this;
     }
 
+    public function setReconnectSuccessCallback(?callable $callback): self
+    {
+        $this->reconnectSuccessCallback = $callback;
+
+        return $this;
+    }
+
     public function consume(Queue $queue, callable $messageCallback, callable $successCallback, callable $errorCallback): void
     {
         $reconnectBackoffMs = self::RECONNECT_BACKOFF_MS;
@@ -42,6 +53,10 @@ public function consume(Queue $queue, callable $messageCallback, callable $succe
              */
             try {
                 $nextMessage = $this->connection->rightPopArray("{$queue->namespace}.queue.{$queue->name}", self::POP_TIMEOUT);
+                if ($reconnectAttempt > 0) {
+                    $this->triggerReconnectSuccessCallback($queue, $reconnectAttempt);
+                }
+
                 $reconnectBackoffMs = self::RECONNECT_BACKOFF_MS;
                 $reconnectAttempt = 0;
             } catch (\RedisException|\RedisClusterException $e) {
@@ -147,6 +162,18 @@ private function triggerReconnectCallback(Queue $queue, \Throwable $error, int $
         }
     }
 
+    private function triggerReconnectSuccessCallback(Queue $queue, int $attempts): void
+    {
+        if (!\is_callable($this->reconnectSuccessCallback)) {
+            return;
+        }
+
+        try {
+            ($this->reconnectSuccessCallback)($queue, $attempts);
+        } catch (\Throwable) {
+        }
+    }
+
     public function enqueue(Queue $queue, array $payload, bool $priority = false): bool
     {
         $payload = [

tests/Queue/E2E/Adapter/RedisReconnectCallbackTest.php

@@ -43,6 +43,36 @@ public function testReconnectCallbackReceivesAttemptContext(): void
         $this->assertGreaterThanOrEqual(0, $calls[0]['sleepMs']);
         $this->assertLessThanOrEqual(100, $calls[0]['sleepMs']);
     }
+
+    public function testReconnectSuccessCallbackReceivesAttemptCount(): void
+    {
+        $queue = new Queue('reconnect-success-callback');
+        $connection = new RecoveringRedisConnection();
+        $broker = new RedisBroker($connection);
+        $calls = [];
+
+        $broker->setReconnectCallback(fn () => null);
+        $broker->setReconnectSuccessCallback(function (Queue $queue, int $attempts) use (&$calls, $broker): void {
+            $calls[] = [
+                'queue' => $queue,
+                'attempts' => $attempts,
+            ];
+
+            $broker->close();
+        });
+
+        $broker->consume(
+            $queue,
+            fn () => null,
+            fn () => null,
+            fn () => null,
+        );
+
+        $this->assertSame(2, $connection->popAttempts);
+        $this->assertCount(1, $calls);
+        $this->assertSame($queue, $calls[0]['queue']);
+        $this->assertSame(1, $calls[0]['attempts']);
+    }
 }
 
 class FailingRedisConnection implements Connection
@@ -160,3 +190,17 @@ public function close(): void
     {
     }
 }
+
+class RecoveringRedisConnection extends FailingRedisConnection
+{
+    public function rightPopArray(string $queue, int $timeout): array|false
+    {
+        $this->popAttempts++;
+
+        if ($this->popAttempts === 1) {
+            throw new \RedisException('Redis is unavailable.');
+        }
+
+        return false;
+    }
+}