initial

Author: iamfarhad

Config/RabbitMQConnectionConfig.php

@@ -0,0 +1,29 @@
+<?php
+
+return [
+    'driver' => 'rabbitmq',
+    'queue' => env('RABBITMQ_QUEUE', 'default'),
+
+    'hosts' => [
+        [
+            'host' => env('RABBITMQ_HOST', '127.0.0.1'),
+            'port' => env('RABBITMQ_PORT', 5672),
+            'user' => env('RABBITMQ_USER', 'guest'),
+            'password' => env('RABBITMQ_PASSWORD', 'guest'),
+            'vhost' => env('RABBITMQ_VHOST', '/'),
+        ],
+    ],
+
+    'options' => [
+        'ssl_options' => [
+            'cafile' => env('RABBITMQ_SSL_CAFILE', null),
+            'local_cert' => env('RABBITMQ_SSL_LOCALCERT', null),
+            'local_key' => env('RABBITMQ_SSL_LOCALKEY', null),
+            'verify_peer' => env('RABBITMQ_SSL_VERIFY_PEER', true),
+            'passphrase' => env('RABBITMQ_SSL_PASSPHRASE', null),
+        ],
+        'queue' => [
+            'job' => \iamfarhad\LaravelRabbitMQ\Jobs\RabbitMQJob::class,
+        ],
+    ],
+];

composer.json

@@ -15,12 +15,19 @@
     },
     "autoload": {
         "psr-4": {
-            "Iamfarhad\\LaravelRabbitMQ\\": "src/"
+            "iamfarhad\\LaravelRabbitMQ\\": "src/"
         }
     },
     "autoload-dev": {
         "psr-4": {
-            "Iamfarhad\\LaravelRabbitMQ\\Tests\\Unit\\": "tests/Unit"
+            "iamfarhad\\LaravelRabbitMQ\\Tests\\Unit\\": "tests/Unit"
+        }
+    },
+    "extra": {
+        "laravel": {
+            "providers": [
+                "iamfarhad\\LaravelRabbitMQ\\LaravelRabbitQueueServiceProvider"
+            ]
         }
     },
     "license": "MIT",
@@ -30,5 +37,8 @@
             "email": "farhad.pd@gmail.com"
         }
     ],
-    "minimum-stability": "dev"
+    "minimum-stability": "dev",
+    "suggest": {
+        "ext-pcntl": "Required to use all features of the queue consumer."
+    }
 }

composer.lock

@@ -0,0 +1,38 @@
+<?php
+
+namespace iamfarhad\LaravelRabbitMQ\Connectors;
+
+use iamfarhad\LaravelRabbitMQ\RabbitQueue;
+use Illuminate\Contracts\Events\Dispatcher;
+use Illuminate\Contracts\Queue\Queue;
+use Illuminate\Queue\Connectors\ConnectorInterface;
+use Illuminate\Queue\Events\WorkerStopping;
+use PhpAmqpLib\Connection\AMQPConnectionConfig;
+use PhpAmqpLib\Connection\AMQPConnectionFactory;
+
+class RabbitMQConnector implements ConnectorInterface
+{
+    private readonly Dispatcher $dispatcher;
+
+    public function __construct(Dispatcher $dispatcher)
+    {
+        $this->dispatcher = $dispatcher;
+    }
+
+    public function connect(array $config = []): Queue
+    {
+        $configs = new AMQPConnectionConfig();
+        // @todo move to config
+        $configs->setHost('eagle-rmq');
+
+        $connection = AMQPConnectionFactory::create($configs);
+
+        $queueConnection = new RabbitQueue($connection, 'default');
+
+        $this->dispatcher->listen(WorkerStopping::class, static function () use ($queueConnection): void {
+            $queueConnection->close();
+        });
+
+        return $queueConnection;
+    }
+}

src/Connectors/RabbitMQConnector.php

@@ -0,0 +1,66 @@
+<?php
+
+namespace iamfarhad\LaravelRabbitMQ\Console;
+
+use Illuminate\Queue\Console\WorkCommand;
+use Illuminate\Support\Str;
+use iamfarhad\LaravelRabbitMQ\Consumer;
+use Symfony\Component\Console\Attribute\AsCommand;
+
+#[AsCommand(name: 'rabbitmq:consume')]
+class ConsumeCommand extends WorkCommand
+{
+    protected $signature = 'rabbitmq:consume
+                            {connection? : The name of the queue connection to work}
+                            {--name=default : The name of the consumer}
+                            {--queue= : The names of the queues to work}
+                            {--once : Only process the next job on the queue}
+                            {--stop-when-empty : Stop when the queue is empty}
+                            {--delay=0 : The number of seconds to delay failed jobs (Deprecated)}
+                            {--backoff=0 : The number of seconds to wait before retrying a job that encountered an uncaught exception}
+                            {--max-jobs=0 : The number of jobs to process before stopping}
+                            {--max-time=0 : The maximum number of seconds the worker should run}
+                            {--force : Force the worker to run even in maintenance mode}
+                            {--memory=128 : The memory limit in megabytes}
+                            {--sleep=3 : Number of seconds to sleep when no job is available}
+                            {--timeout=60 : The number of seconds a child process can run}
+                            {--tries=1 : Number of times to attempt a job before logging it failed}
+                            {--rest=0 : Number of seconds to rest between jobs}
+                            {--max-priority=}
+                            {--consumer-tag}
+                            {--prefetch-size=0}
+                            {--prefetch-count=1000}
+                           ';
+
+    protected $description = 'Consume messages';
+
+    public function handle(): void
+    {
+        /** @var Consumer $consumer */
+        $consumer = $this->worker;
+
+        $consumer->setContainer($this->laravel);
+        $consumer->setName($this->option('name'));
+        $consumer->setConsumerTag($this->consumerTag());
+        $consumer->setMaxPriority((int) $this->option('max-priority'));
+        $consumer->setPrefetchSize((int) $this->option('prefetch-size'));
+        $consumer->setPrefetchCount((int) $this->option('prefetch-count'));
+
+        parent::handle();
+    }
+
+    protected function consumerTag(): string
+    {
+        if ($consumerTag = $this->option('consumer-tag')) {
+            return $consumerTag;
+        }
+
+        $consumerTag = implode('_', [
+            Str::slug(config('app.name', 'laravel')),
+            Str::slug($this->option('name')),
+            md5(serialize($this->options()) . Str::random(16) . getmypid()),
+        ]);
+
+        return Str::substr($consumerTag, 0, 255);
+    }
+}