Merge pull request vyuldashev#154 from formapro-forks/amqp-interop

Amqp interop based version.

Author: vyuldashev

README.md

@@ -11,19 +11,15 @@ RabbitMQ Queue driver for Laravel
 1. Install this package via composer using:
 
 ```
-composer require vladimir-yuldashev/laravel-queue-rabbitmq:5.5
+composer require vladimir-yuldashev/laravel-queue-rabbitmq:^6
 ```
 
 2. Add these properties to `.env` with proper values:
 
 ```
 QUEUE_DRIVER=rabbitmq
 
-RABBITMQ_HOST=127.0.0.1
-RABBITMQ_PORT=5672
-RABBITMQ_VHOST=/
-RABBITMQ_LOGIN=guest
-RABBITMQ_PASSWORD=guest
+RABBITMQ_DSN=amqp://guest:guest@127.0.0.1:5672/%2F
 RABBITMQ_QUEUE=queue_name
 ```
 

composer.json

@@ -13,17 +13,23 @@
     "illuminate/database": "5.5.*",
     "illuminate/support": "5.5.*",
     "illuminate/queue": "5.5.*",
-    "php-amqplib/php-amqplib": "2.6.*"
+    "enqueue/amqp-lib": "0.8.5"
   },
   "require-dev": {
     "phpunit/phpunit": "~6.0",
+    "illuminate/events": "5.5.*",
     "mockery/mockery": "^0.9.5"
   },
   "autoload": {
     "psr-4": {
       "VladimirYuldashev\\LaravelQueueRabbitMQ\\": "src/"
     }
   },
+  "autoload-dev": {
+    "psr-4": {
+      "VladimirYuldashev\\LaravelQueueRabbitMQ\\Tests\\": "tests/"
+    }
+  },
   "extra": {
     "laravel": {
       "providers": [

config/rabbitmq.php

@@ -9,6 +9,16 @@
 
     'driver' => 'rabbitmq',
 
+    'dsn' => env('RABBITMQ_DSN', null),
+
+    /*
+     * Could be one a class that implements \Interop\Amqp\AmqpConnectionFactory for example:
+     *  - \EnqueueAmqpExt\AmqpConnectionFactory if you install enqueue/amqp-ext
+     *  - \EnqueueAmqpLib\AmqpConnectionFactory if you install enqueue/amqp-lib
+     *  - \EnqueueAmqpBunny\AmqpConnectionFactory if you install enqueue/amqp-bunny
+     */
+    'factory_class' => \Enqueue\AmqpLib\AmqpConnectionFactory::class,
+
     'host' => env('RABBITMQ_HOST', '127.0.0.1'),
     'port' => env('RABBITMQ_PORT', 5672),
 
@@ -26,6 +36,11 @@
      */
     'exchange_declare' => env('RABBITMQ_EXCHANGE_DECLARE', true),
 
+    /*
+     * Determine if queue should be created if it does not exist.
+     */
+    'queue_declare' => env('RABBITMQ_QUEUE_DECLARE', true),
+
     /*
      * Determine if queue should be created and binded to the exchange if it does not exist.
      */
@@ -43,7 +58,7 @@
     ],
     'exchange_params' => [
         'name' => env('RABBITMQ_EXCHANGE_NAME'),
-        'type' => env('RABBITMQ_EXCHANGE_TYPE', 'direct'),
+        'type' => env('RABBITMQ_EXCHANGE_TYPE', \Interop\Amqp\AmqpTopic::TYPE_DIRECT),
         'passive' => env('RABBITMQ_EXCHANGE_PASSIVE', false),
         'durable' => env('RABBITMQ_EXCHANGE_DURABLE', true),
         'auto_delete' => env('RABBITMQ_EXCHANGE_AUTODELETE', false),
@@ -62,6 +77,7 @@
         'ssl_on'        => env('RABBITMQ_SSL', false),
         '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),
     ]

src/LaravelQueueRabbitMQServiceProvider.php

@@ -4,9 +4,7 @@
 
 use Illuminate\Queue\QueueManager;
 use Illuminate\Support\ServiceProvider;
-use PhpAmqpLib\Connection\AMQPStreamConnection;
 use VladimirYuldashev\LaravelQueueRabbitMQ\Queue\Connectors\RabbitMQConnector;
-use VladimirYuldashev\LaravelQueueRabbitMQ\Queue\Connectors\RabbitMQConnectorSSL;
 
 class LaravelQueueRabbitMQServiceProvider extends ServiceProvider
 {
@@ -32,20 +30,8 @@ public function boot()
         /** @var QueueManager $queue */
         $queue = $this->app['queue'];
 
-        if ($this->app['config']['rabbitmq']['ssl_params']['ssl_on'] === true) {
-            $connector = new RabbitMQConnectorSSL();
-        } else {
-            $connector = new RabbitMQConnector();
-        }
-
-        $queue->stopping(function () use ($connector) {
-            if ($connector->connection() instanceof AMQPStreamConnection) {
-                $connector->connection()->close();
-            }
-        });
-
-        $queue->addConnector('rabbitmq', function () use ($connector) {
-            return $connector;
+        $queue->addConnector('rabbitmq', function () {
+            return new RabbitMQConnector($this->app['events']);
         });
     }
 }

src/Queue/Connectors/RabbitMQConnector.php

@@ -2,15 +2,27 @@
 
 namespace VladimirYuldashev\LaravelQueueRabbitMQ\Queue\Connectors;
 
+use Enqueue\AmqpTools\DelayStrategyAware;
+use Enqueue\AmqpTools\RabbitMqDlxDelayStrategy;
+use Illuminate\Contracts\Events\Dispatcher;
 use Illuminate\Contracts\Queue\Queue;
 use Illuminate\Queue\Connectors\ConnectorInterface;
-use PhpAmqpLib\Connection\AMQPStreamConnection;
+use Illuminate\Queue\Events\WorkerStopping;
+use Interop\Amqp\AmqpConnectionFactory as InteropAmqpConnectionFactory;
+use Interop\Amqp\AmqpConnectionFactory;
 use VladimirYuldashev\LaravelQueueRabbitMQ\Queue\RabbitMQQueue;
 
 class RabbitMQConnector implements ConnectorInterface
 {
-    /** @var AMQPStreamConnection */
-    private $connection;
+    /**
+     * @var Dispatcher
+     */
+    private $dispatcher;
+
+    public function __construct(Dispatcher $dispatcher)
+    {
+        $this->dispatcher = $dispatcher;
+    }
 
     /**
      * Establish a queue connection.
@@ -21,23 +33,41 @@ class RabbitMQConnector implements ConnectorInterface
      */
     public function connect(array $config): Queue
     {
-        // create connection with AMQP
-        $this->connection = new AMQPStreamConnection(
-            $config['host'],
-            $config['port'],
-            $config['login'],
-            $config['password'],
-            $config['vhost']
-        );
-
-        return new RabbitMQQueue(
-            $this->connection,
-            $config
-        );
-    }
+        if (false == array_key_exists('factory_class', $config)) {
+            throw new \LogicException('The factory_class option is missing though it is required.');
+        }
 
-    public function connection()
-    {
-        return $this->connection;
+        $factoryClass = $config['factory_class'];
+        if (false == class_exists($factoryClass) || false == (new \ReflectionClass($factoryClass))->implementsInterface(InteropAmqpConnectionFactory::class)) {
+            throw new \LogicException(sprintf('The factory_class option has to be valid class that implements "%s"', InteropAmqpConnectionFactory::class));
+        }
+
+        /** @var AmqpConnectionFactory $factory */
+        $factory = new $factoryClass([
+            'dsn' => $config['dsn'],
+            'host' => $config['host'],
+            'port' => $config['port'],
+            'user' => $config['login'],
+            'pass' => $config['password'],
+            'vhost' => $config['vhost'],
+            'ssl_on' => $config['ssl_params']['ssl_on'],
+            'ssl_verify' => $config['ssl_params']['verify_peer'],
+            'ssl_cacert' => $config['ssl_params']['cafile'],
+            'ssl_cert' => $config['ssl_params']['local_cert'],
+            'ssl_key' => $config['ssl_params']['local_key'],
+            'ssl_passphrase' => $config['ssl_params']['passphrase'],
+        ]);
+
+        if ($factory instanceof DelayStrategyAware) {
+            $factory->setDelayStrategy(new RabbitMqDlxDelayStrategy());
+        }
+
+        $context = $factory->createContext();
+
+        $this->dispatcher->listen(WorkerStopping::class, function () use ($context) {
+            $context->close();
+        });
+
+        return new RabbitMQQueue($context, $config);
     }
 }

src/Queue/Connectors/RabbitMQConnectorSSL.php

@@ -9,8 +9,8 @@
 use Illuminate\Queue\Jobs\Job;
 use Illuminate\Queue\Jobs\JobName;
 use Illuminate\Support\Str;
-use PhpAmqpLib\Channel\AMQPChannel;
-use PhpAmqpLib\Message\AMQPMessage;
+use Interop\Amqp\AmqpConsumer;
+use Interop\Amqp\AmqpMessage;
 use VladimirYuldashev\LaravelQueueRabbitMQ\Queue\RabbitMQQueue;
 
 class RabbitMQJob extends Job implements JobContract
@@ -23,23 +23,20 @@ class RabbitMQJob extends Job implements JobContract
     const ATTEMPT_COUNT_HEADERS_KEY = 'attempts_count';
 
     protected $connection;
-    protected $channel;
+    protected $consumer;
     protected $message;
 
     public function __construct(
         Container $container,
         RabbitMQQueue $connection,
-        AMQPChannel $channel,
-        string $queue,
-        AMQPMessage $message,
-        string $connectionName = null
+        AmqpConsumer $consumer,
+        AmqpMessage $message
     ) {
         $this->container = $container;
         $this->connection = $connection;
-        $this->channel = $channel;
-        $this->queue = $queue;
+        $this->consumer = $consumer;
         $this->message = $message;
-        $this->connectionName = $connectionName;
+        $this->queue = $consumer->getQueue()->getQueueName();
     }
 
     /**
@@ -79,16 +76,10 @@ public function fire()
      */
     public function attempts(): int
     {
-        if ($this->message->has('application_headers') === true) {
-            $headers = $this->message->get('application_headers')->getNativeData();
-
-            if (isset($headers[self::ATTEMPT_COUNT_HEADERS_KEY]) === true) {
-                return $headers[self::ATTEMPT_COUNT_HEADERS_KEY];
-            }
-        }
-
         // set default job attempts to 1 so that jobs can run without retry
-        return 1;
+        $defaultAttempts = 1;
+
+        return $this->message->getProperty(self::ATTEMPT_COUNT_HEADERS_KEY, $defaultAttempts);
     }
 
     /**
@@ -98,14 +89,15 @@ public function attempts(): int
      */
     public function getRawBody(): string
     {
-        return $this->message->body;
+        return $this->message->getBody();
     }
 
     /** @inheritdoc */
     public function delete()
     {
         parent::delete();
-        $this->channel->basic_ack($this->message->get('delivery_tag'));
+
+        $this->consumer->acknowledge($this->message);
     }
 
     /** @inheritdoc */
@@ -143,7 +135,7 @@ public function release($delay = 0)
      */
     public function getJobId(): string
     {
-        return $this->message->get('correlation_id');
+        return $this->message->getCorrelationId();
     }
 
     /**