amqp interop based version.

Author: makasim

composer.json

@@ -13,7 +13,7 @@
     "illuminate/database": "5.5.*",
     "illuminate/support": "5.5.*",
     "illuminate/queue": "5.5.*",
-    "php-amqplib/php-amqplib": "2.6.*"
+    "enqueue/amqp-lib": "^0.8"
   },
   "require-dev": {
     "phpunit/phpunit": "~6.0",

src/Queue/Connectors/RabbitMQConnector.php

@@ -2,16 +2,14 @@
 
 namespace VladimirYuldashev\LaravelQueueRabbitMQ\Queue\Connectors;
 
+use Enqueue\AmqpLib\AmqpConnectionFactory;
+use Enqueue\AmqpTools\RabbitMqDlxDelayStrategy;
 use Illuminate\Contracts\Queue\Queue;
 use Illuminate\Queue\Connectors\ConnectorInterface;
-use PhpAmqpLib\Connection\AMQPStreamConnection;
 use VladimirYuldashev\LaravelQueueRabbitMQ\Queue\RabbitMQQueue;
 
 class RabbitMQConnector implements ConnectorInterface
 {
-    /** @var AMQPStreamConnection */
-    private $connection;
-
     /**
      * Establish a queue connection.
      *
@@ -21,23 +19,19 @@ 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']
-        );
+        $factory = new AmqpConnectionFactory([
+            'host' => $config['host'],
+            'port' => $config['port'],
+            'user' => $config['login'],
+            'pass' => $config['password'],
+            'vhost' => $config['vhost'],
+        ]);
+
+        $factory->setDelayStrategy(new RabbitMqDlxDelayStrategy());
 
         return new RabbitMQQueue(
-            $this->connection,
+            $factory->createContext(),
             $config
         );
     }
-
-    public function connection()
-    {
-        return $this->connection;
-    }
 }

src/Queue/Jobs/RabbitMQJob.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,20 +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,
+        AmqpConsumer $consumer,
         string $queue,
-        AMQPMessage $message,
+        AmqpMessage $message,
         string $connectionName = null
     ) {
         $this->container = $container;
         $this->connection = $connection;
-        $this->channel = $channel;
+        $this->consumer = $consumer;
         $this->queue = $queue;
         $this->message = $message;
         $this->connectionName = $connectionName;
@@ -79,16 +79,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 +92,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 +138,7 @@ public function release($delay = 0)
      */
     public function getJobId(): string
     {
-        return $this->message->get('correlation_id');
+        return $this->message->getCorrelationId();
     }
 
     /**