laravel 5.7 compatibility fixes

posix recommendation

Author: martianoff

composer.json

@@ -5,18 +5,21 @@
   "license": "GPL-2.0",
   "require": {
     "php": ">=7.1",
-    "php-amqplib/php-amqplib": "2.6.*"
+    "php-amqplib/php-amqplib": "2.6.*",
+    "ext-json": "*"
   },
   "require-dev": {
-    "laravel/framework": ">=5.3.0",
+    "laravel/framework": ">=5.7.0",
     "squizlabs/php_codesniffer": "1.*",
-    "orchestra/testbench": "~3.4",
-    "phpunit/phpunit": "~6.1"
+    "orchestra/testbench": "3.8.*"
   },
   "autoload": {
     "psr-4": {
       "Forumhouse\\LaravelAmqp\\": "src/",
       "Forumhouse\\LaravelAmqp\\Tests\\": "tests/"
     }
+  },
+  "suggest": {
+    "ext-posix": "Restart workers if connection is lost to avoid unlimited loop"
   }
 }

src/Queue/AMQPQueue.php

@@ -9,7 +9,6 @@
 use Illuminate\Queue\InvalidPayloadException;
 use Illuminate\Queue\Queue;
 use PhpAmqpLib\Channel\AMQPChannel;
-use PhpAmqpLib\Connection\AMQPConnection;
 use PhpAmqpLib\Connection\AMQPStreamConnection;
 use PhpAmqpLib\Message\AMQPMessage;
 use PhpAmqpLib\Wire\AMQPTable;
@@ -30,7 +29,7 @@ class AMQPQueue extends Queue implements QueueContract
     const EXCHANGE_TYPE_FANOUT = 'fanout';
 
     /**
-     * @var AMQPConnection Connection to amqp compatible server
+     * @var AMQPStreamConnection Connection to amqp compatible server
      */
     protected $connection;
 
@@ -164,7 +163,7 @@ public function getCustomMessageOptions(){
     public function push($job, $data = '', $queue = null)
     {
         $queue = $this->prepareQueue($queue);
-        $payload = new AMQPMessage($this->createPayload($job, $data), $this->messageProperties);
+        $payload = new AMQPMessage($this->createPayload($job, $queue, $data), $this->messageProperties);
         $this->channel->basic_publish($payload, $this->exchangeName, $this->getRoutingKey($queue));
 
         return true;
@@ -185,7 +184,7 @@ public function push($job, $data = '', $queue = null)
     public function addMessageToBatch($job, $data = '', $queue = null)
     {
         $queue = $this->prepareQueue($queue);
-        $payload = new AMQPMessage($this->createPayload($job, $data), $this->messageProperties);
+        $payload = new AMQPMessage($this->createPayload($job, $queue, $data), $this->messageProperties);
         $this->channel->batch_basic_publish($payload, $this->exchangeName, $this->getRoutingKey($queue));
 
         return true;
@@ -313,7 +312,7 @@ public function later($delay, $job, $data = '', $queue = null)
         $queue = $this->prepareQueue($queue);
         $delayedQueueName = $this->declareDelayedQueue($queue, $delay);
 
-        $payload = new AMQPMessage($this->createPayload($job, $data), $this->messageProperties);
+        $payload = new AMQPMessage($this->createPayload($job, $queue, $data), $this->messageProperties);
         $this->channel->basic_publish($payload, $this->exchangeName, $delayedQueueName);
         return true;
     }
@@ -431,15 +430,16 @@ private function prepareQueue($queue)
      * Create a payload string from the given job and data.
      *
      * @param  string  $job
+     * @param  string  $queue
      * @param  mixed   $data
      * @return string
      *
      * @throws \Illuminate\Queue\InvalidPayloadException
      */
-    protected function createPayload($job, $data = '')
+    protected function createPayload($job, $queue, $data = '')
     {
         $data = is_array($data) ? array_merge($data, $this->getCustomMessageOptions()) : $this->getCustomMessageOptions();
-        $payload = json_encode($this->createPayloadArray($job, $data));
+        $payload = json_encode($this->createPayloadArray($job, $queue, $data));
         if (JSON_ERROR_NONE !== json_last_error()) {
             throw new InvalidPayloadException(
                 'Unable to JSON encode payload. Error code: '.json_last_error()
@@ -453,20 +453,28 @@ protected function createPayload($job, $data = '')
      * Create a payload for an object-based queue handler.
      *
      * @param  mixed  $job
+     * @param  string  $queue
      * @return array
      */
-    protected function createObjectPayload($job)
+    protected function createObjectPayload($job, $queue)
     {
-        return [
+        $payload = $this->withCreatePayloadHooks($queue, [
             'displayName' => $this->getDisplayName($job),
             'job' => 'Illuminate\Queue\CallQueuedHandler@call',
             'maxTries' => $job->tries ?? null,
             'timeout' => $job->timeout ?? null,
             'timeoutAt' => $this->getJobExpiration($job),
+            'data' => [
+                'commandName' => $job,
+                'command' => $job,
+            ],
+        ]);
+
+        return array_merge($payload, [
             'data' => array_merge([
                 'commandName' => get_class($job),
                 'command' => serialize(clone $job),
             ],$this->getCustomMessageOptions()),
-        ];
+        ]);
     }
 }

src/Traits/AMQPFailureDetector.php

@@ -9,7 +9,7 @@ trait AMQPFailureDetector
 
     public function catchAMQPConnectionFailure(Exception $exception)
     {
-        if(\App::runningInConsole() && $this->causedByLostConnection($exception) && extension_loaded('posix')){
+        if(app()->runningInConsole() && $this->causedByLostConnection($exception) && extension_loaded('posix')){
             posix_kill(getmypid(), SIGTERM);
         }
     }

tests/BaseFeaturesTest.php

@@ -18,15 +18,6 @@ class BaseFeaturesTest extends LaravelAmqpTestBase
      */
     const TEST_JOB_CLASS = 'Forumhouse\LaravelAmqp\Tests\TestJobHandler';
 
-    /**
-     *
-     */
-    public function setUp()
-    {
-        parent::setUp();
-    }
-
-
     /**
      * Test for a job, ending with a delete() call (deleted from queue)
      */