feat(publisher): add ping method to publisher

src/Queue/Broker/AMQP.php

@@ -128,6 +128,11 @@ public function close(): void
         }
     }
 
+    public function ping(): bool
+    {
+        return $this->withChannel(fn (AMQPChannel $channel) => $channel->is_open());
+    }
+
     public function enqueue(Queue $queue, array $payload): bool
     {
         $payload = [
@@ -172,10 +177,10 @@ public function getQueueSize(Queue $queue, bool $failedJobs = false): int
     }
 
     /**
-     * @param callable(AMQPChannel $channel): void $callback
+     * @param callable(AMQPChannel $channel): mixed $callback
      * @throws \Exception
      */
-    private function withChannel(callable $callback): void
+    private function withChannel(callable $callback): mixed
     {
         $createChannel = function (): AMQPChannel {
             $connection = new AMQPStreamConnection($this->host, $this->port, $this->user, $this->password, $this->vhost, heartbeat: $this->heartbeat);
@@ -194,13 +199,13 @@ private function withChannel(callable $callback): void
         }
 
         try {
-            $callback($this->channel);
+            return $callback($this->channel);
         } catch (\Throwable $th) {
             // createChannel() might throw, in that case set the channel to `null` first.
             $this->channel = null;
             // try creating a new connection once, if this still fails, throw the error
             $this->channel = $createChannel();
-            $callback($this->channel);
+            return $callback($this->channel);
         }
     }
 }

src/Queue/Broker/Redis.php

@@ -94,6 +94,11 @@ public function close(): void
         $this->closed = true;
     }
 
+    public function ping(): bool
+    {
+        return $this->connection->ping();
+    }
+
     public function enqueue(Queue $queue, array $payload): bool
     {
         $payload = [

src/Queue/Publisher.php

@@ -4,6 +4,13 @@
 
 interface Publisher
 {
+    /**
+     * Checks if the publisher can reach the queue.
+     *
+     * @return bool
+     */
+    public function ping(): bool;
+
     /**
      * Publishes a new message onto the queue.
      *