refactor: simplify README and enhance RabbitMQ consumer implementation

- Streamlined the README by reducing verbosity and improving clarity, including a new Table of Contents.
- Removed the ConsumerInterface and integrated its functionality directly into the Consumer class.
- Enhanced error handling in RabbitMQ connection and message publishing methods.
- Updated job handling in RabbitMQJob to ensure better logging and connection checks.
- Improved the RabbitQueue class with retry logic and better exception handling for queue operations.

Author: farhadzand

README.md

@@ -1,8 +1,12 @@
 <?php
 
+declare(strict_types=1);
+
 namespace iamfarhad\LaravelRabbitMQ\Connectors;
 
 use AMQPConnection;
+use AMQPConnectionException;
+use iamfarhad\LaravelRabbitMQ\Exceptions\ConnectionException;
 use iamfarhad\LaravelRabbitMQ\RabbitQueue;
 use Illuminate\Contracts\Events\Dispatcher;
 use Illuminate\Contracts\Queue\Queue;
@@ -13,6 +17,9 @@
 {
     public function __construct(private Dispatcher $dispatcher) {}
 
+    /**
+     * @throws ConnectionException
+     */
     public function connect(array $config = []): Queue
     {
         $connectionConfig = [
@@ -41,8 +48,16 @@ public function connect(array $config = []): Queue
         }
 
         // Create AMQP Connection
-        $connection = new AMQPConnection($connectionConfig);
-        $connection->connect();
+        try {
+            $connection = new AMQPConnection($connectionConfig);
+            $connection->connect();
+        } catch (AMQPConnectionException $e) {
+            throw new ConnectionException(
+                'Failed to connect to RabbitMQ: '.$e->getMessage(),
+                $e->getCode(),
+                $e
+            );
+        }
 
         $defaultQueue = config('queue.connections.rabbitmq.queue', 'default');
         $options = config('queue.connections.rabbitmq.options', []);

src/Connectors/RabbitMQConnector.php

@@ -88,7 +88,6 @@ public function handle(): int
             // Parent process
             $childPids[] = $pid;
             $this->info("Started worker process $pid");
-
         }
 
         // Set up signal handling for graceful termination
@@ -132,7 +131,7 @@ private function consume(): int
             }
 
             $consumer->setContainer($this->laravel);
-            $consumer->setName($this->option('name'));
+            $consumer->setName((string) $this->option('name'));
             $consumer->setConsumerTag($this->generateConsumerTag());
 
             // Only set max priority if it's provided and not null
@@ -154,8 +153,9 @@ private function consume(): int
      */
     private function generateConsumerTag(): string
     {
-        if ($consumerTag = $this->option('consumer-tag')) {
-            return $consumerTag;
+        $consumerTag = $this->option('consumer-tag');
+        if ($consumerTag !== null && $consumerTag !== false && $consumerTag !== '') {
+            return (string) $consumerTag;
         }
 
         $appName = config('app.name', 'laravel');
@@ -166,7 +166,7 @@ private function generateConsumerTag(): string
             '_',
             [
                 Str::slug($appName),
-                Str::slug($consumerName),
+                Str::slug((string) $consumerName),
                 $uniqueId,
             ]
         );

src/Console/ConsumeCommand.php

@@ -1,29 +1,33 @@
 <?php
 
+declare(strict_types=1);
+
 namespace iamfarhad\LaravelRabbitMQ;
 
 use AMQPChannel;
 use AMQPChannelException;
 use AMQPConnectionException;
+use AMQPEnvelope;
 use AMQPQueue;
 use Exception;
-use iamfarhad\LaravelRabbitMQ\Contracts\ConsumerInterface;
+use iamfarhad\LaravelRabbitMQ\Jobs\RabbitMQJob;
 use Illuminate\Container\Container;
 use Illuminate\Queue\Worker;
 use Illuminate\Queue\WorkerOptions;
+use RuntimeException;
 use Throwable;
 
-class Consumer extends Worker implements ConsumerInterface
+class Consumer extends Worker
 {
     private Container $container;
 
-    private string $consumerTag;
+    private string $consumerTag = '';
 
-    private int $maxPriority;
+    private int $maxPriority = 0;
 
     private AMQPChannel $amqpChannel;
 
-    private ?object $currentJob = null;
+    private ?RabbitMQJob $currentJob = null;
 
     public function setContainer(Container $container): void
     {
@@ -63,7 +67,7 @@ public function daemon($connectionName, $queue, WorkerOptions $options)
 
         // Check if the connection is a RabbitQueue instance
         if (! $connection instanceof RabbitQueue) {
-            throw new \RuntimeException('Connection must be an instance of RabbitQueue for RabbitMQ Consumer');
+            throw new RuntimeException('Connection must be an instance of RabbitQueue for RabbitMQ Consumer');
         }
 
         $connection->declareQueue($queue);
@@ -91,7 +95,7 @@ public function daemon($connectionName, $queue, WorkerOptions $options)
                 // Try to get a message from the queue
                 $envelope = $amqpQueue->get(AMQP_NOPARAM);
 
-                if ($envelope !== false) {
+                if ($envelope instanceof AMQPEnvelope) {
                     $job = new $jobClass(
                         $this->container,
                         $connection,
@@ -100,6 +104,7 @@ public function daemon($connectionName, $queue, WorkerOptions $options)
                         $queue
                     );
 
+                    /** @var RabbitMQJob $job */
                     $this->currentJob = $job;
 
                     if ($this->supportsAsyncSignals()) {

src/Consumer.php

@@ -1,8 +1,10 @@
 <?php
 
+declare(strict_types=1);
+
 namespace iamfarhad\LaravelRabbitMQ\Exceptions;
 
-class ConnectionException extends RabbitMQException
+final class ConnectionException extends RabbitMQException
 {
     //
 }

src/Contracts/ConsumerInterface.php

@@ -1,8 +1,10 @@
 <?php
 
+declare(strict_types=1);
+
 namespace iamfarhad\LaravelRabbitMQ\Exceptions;
 
-class QueueException extends RabbitMQException
+final class QueueException extends RabbitMQException
 {
     //
 }

src/Exceptions/ConnectionException.php

@@ -1,10 +1,12 @@
 <?php
 
+declare(strict_types=1);
+
 namespace iamfarhad\LaravelRabbitMQ\Exceptions;
 
 use Exception;
 
-class RabbitMQException extends Exception
+abstract class RabbitMQException extends Exception
 {
     //
 }

src/Exceptions/QueueException.php

@@ -1,19 +1,22 @@
 <?php
 
+declare(strict_types=1);
+
 namespace iamfarhad\LaravelRabbitMQ\Facades;
 
 use Illuminate\Support\Facades\Facade;
 
 /**
- * @method static mixed push($job, $data = '', $queue = null)
- * @method static mixed pushRaw($payload, $queue = null, array $options = [])
- * @method static mixed later($delay, $job, $data = '', $queue = null)
- * @method static mixed laterRaw($delay, $payload, $queue = null, $attempts = 2)
- * @method static mixed pop($queue = null)
- * @method static int size($queue = null)
+ * @method static ?string push($job, $data = '', ?string $queue = null)
+ * @method static ?string pushRaw(string $payload, ?string $queue = null, array $options = [])
+ * @method static ?string later($delay, $job, $data = '', ?string $queue = null)
+ * @method static ?string laterRaw($delay, string $payload, ?string $queue = null, int $attempts = 2)
+ * @method static ?\Illuminate\Contracts\Queue\Job pop(?string $queue = null)
+ * @method static int size(?string $queue = null)
  * @method static bool queueExists(string $queueName)
  * @method static mixed purgeQueue(string $queueName)
  * @method static mixed deleteQueue(string $queueName)
+ * @method static void declareQueue(string $name, bool $durable = true, bool $autoDelete = false, array $arguments = [])
  *
  * @see \iamfarhad\LaravelRabbitMQ\RabbitQueue
  */

src/Exceptions/RabbitMQException.php

@@ -1,5 +1,7 @@
 <?php
 
+declare(strict_types=1);
+
 namespace iamfarhad\LaravelRabbitMQ\Jobs;
 
 use AMQPEnvelope;
@@ -8,10 +10,14 @@
 use Illuminate\Contracts\Queue\Job as JobContract;
 use Illuminate\Queue\Jobs\Job;
 use Illuminate\Support\Arr;
+use Illuminate\Support\Facades\Log;
 use JsonException;
+use Throwable;
 
 final class RabbitMQJob extends Job implements JobContract
 {
+    private const FAILED_MESSAGES_EXCHANGE = 'failed_messages';
+
     private readonly array $decoded;
 
     public function __construct(
@@ -30,7 +36,7 @@ public function __construct(
     /**
      * {@inheritdoc}
      */
-    public function getJobId()
+    public function getJobId(): ?string
     {
         return $this->decoded['id'] ?? null;
     }
@@ -62,9 +68,28 @@ public function attempts(): int
      */
     private function convertMessageToFailed(): void
     {
-        if ($this->amqpEnvelope->getExchangeName() !== 'failed_messages') {
-            $this->rabbitQueue->declareQueue('failed_messages');
-            $this->rabbitQueue->pushRaw($this->amqpEnvelope->getBody(), 'failed_messages');
+        try {
+            if ($this->amqpEnvelope->getExchangeName() !== self::FAILED_MESSAGES_EXCHANGE) {
+                // Check if the RabbitMQ connection is still available before attempting operations
+                if (! $this->rabbitQueue->getConnection()->isConnected()) {
+                    Log::warning('RabbitMQ connection lost, cannot move job to failed_messages queue', [
+                        'job_id' => $this->getJobId(),
+                    ]);
+
+                    return;
+                }
+
+                $this->rabbitQueue->declareQueue(self::FAILED_MESSAGES_EXCHANGE);
+                $this->rabbitQueue->pushRaw($this->amqpEnvelope->getBody(), self::FAILED_MESSAGES_EXCHANGE);
+            }
+        } catch (Throwable $e) {
+            // If channel is not available or queue declaration fails, just log the error
+            // This can happen when the connection is lost during job processing
+            Log::error('Failed to move job to failed_messages queue', [
+                'job_id' => $this->getJobId(),
+                'exception' => $e->getMessage(),
+                'exception_class' => get_class($e),
+            ]);
         }
     }