require php-amqplib/php-amqplib >= 3.0

Author: jooola

composer.json

@@ -30,7 +30,7 @@
     },
     "require-dev": {
         "ext-amqp": "*",
-        "php-amqplib/php-amqplib": "*",
+        "php-amqplib/php-amqplib": ">=3.0",
         "phpunit/phpunit": "<6.0.0",
         "predis/predis": "*"
     },

src/AMQPLibConnector.php

@@ -1,4 +1,5 @@
 <?php
+
 /**
  * This file contains a PHP client to Celery distributed task queue
  *
@@ -67,7 +68,7 @@ class AMQPLibConnector extends AbstractAMQPConnector
 
     public function GetConnectionObject($details)
     {
-        return new \PhpAmqpLib\Connection\AMQPConnection(
+        return new \PhpAmqpLib\Connection\AMQPStreamConnection(
             $details['host'],
             $details['port'],
             $details['login'],
@@ -129,14 +130,14 @@ public function Consume($msg)
 
     /**
      * Return result of task execution for $task_id
-     * @param \PhpAmqpLib\Connection\AMQPConnection $connection
+     * @param \PhpAmqpLib\Connection\AMQPStreamConnection $connection
      * @param string $task_id Celery task identifier
      * @param int $expire expire time result queue, milliseconds
      * @param boolean $removeMessageFromQueue whether to remove message from queue
      * @return array array('body' => JSON-encoded message body, 'complete_result' => AMQPMessage object)
      * 			or false if result not ready yet
      */
-    public function GetMessageBody($connection, $task_id, $expire=0, $removeMessageFromQueue = true)
+    public function GetMessageBody($connection, $task_id, $expire = 0, $removeMessageFromQueue = true)
     {
         if (!$this->receiving_channel) {
             $ch = $connection->channel();

src/AbstractAMQPConnector.php

@@ -53,7 +53,7 @@ public static function GetBestInstalledExtensionName($ssl = false)
             return 'php-amqplib-ssl';
         } elseif (class_exists('\AMQPConnection') && extension_loaded('amqp')) {
             return 'pecl';
-        } elseif (class_exists('\PhpAmqpLib\Connection\AMQPConnection')) {
+        } elseif (class_exists('\PhpAmqpLib\Connection\AMQPStreamConnection')) {
             return 'php-amqplib';
         } else {
             return 'unknown';