Merge pull request #26 from fhteam/maksimru-fix/laravel-5.7

Maksimru fix/laravel 5.7

Author: god1dog

README.md

@@ -41,13 +41,54 @@ In your ```config/queue.php``` file you have to provide the following:
         'exchange_name' => null,
         'exchange_type' => null,
         'exchange_flags' => null,
+        'keepalive' > false,
+        'heartbeat' => 0,
+        'retry_after' => 0,
         ),
 ),
 ```
 
 In your ```config/app.php``` add ```'Forumhouse\LaravelAmqp\ServiceProvider\LaravelAmqpServiceProvider'``` to the list of service 
 providers registered.
 
+Improved worker stability (PHP 7.1+ is required)
+------------
+
+For better stability please add following code in app/Exceptions/Handler.php:
+
+```php
+class Handler extends ExceptionHandler
+{
+```
+
+to
+
+```php
+class Handler extends ExceptionHandler
+{
+    use AMQPFailureDetector;
+```
+
+And
+
+```php
+public function report(Exception $exception)
+{
+    parent::report($exception);
+}
+```
+
+to
+
+```php
+public function report(Exception $exception)
+{
+    $this->catchAMQPConnectionFailure($exception);
+    parent::report($exception);
+}
+```
+
+
 Usage
 ------------
 

composer.json

@@ -4,19 +4,22 @@
   "minimum-stability": "stable",
   "license": "GPL-2.0",
   "require": {
-    "php": ">=5.4.0",
-    "php-amqplib/php-amqplib": "2.6.*"
+    "php": ">=7.1",
+    "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/Connectors/AmqpConnector.php

@@ -48,9 +48,9 @@ public function connect(array $config)
         }
 
         return new AMQPQueue(
-            $connection, $config['queue'], $config['queue_flags'], $config['declare_queues'],
+            $connection, $config['queue'], $config['queue_flags'], isset($config['declare_queues']) ? $config['declare_queues'] : true,
             $config['message_properties'], $config['channel_id'],
-            $config['exchange_name'], $config['exchange_type'], $config['exchange_flags']
+            $config['exchange_name'], $config['exchange_type'], $config['exchange_flags'], (isset($config['retry_after']) ? $config['retry_after'] : 0)
         );
     }
 }

src/Jobs/AMQPJob.php

@@ -70,6 +70,10 @@ public function release($delay = 0)
         $body['attempts'] = $this->attempts() + 1;
         $job = $body['job'];
 
+        //if retry_after option is set use it on failure instead of traditional delay
+        if(isset($body['data']['retryAfter']) && $body['data']['retryAfter'] > 0)
+            $delay = $body['data']['retryAfter'];
+
         /** @var QueueContract $queue */
         $queue = $this->container['queue']->connection();
         if ($delay > 0) {
@@ -119,10 +123,14 @@ public function getQueue()
      * Get the job identifier.
      *
      * @return string
-     * @throws \OutOfBoundsException
      */
     public function getJobId()
     {
-        return $this->amqpMessage->get('message_id');
+        try {
+            return $this->amqpMessage->get('message_id');
+        } catch (\OutOfBoundsException $exception){
+            return null;
+        }
     }
+
 }

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;
 
@@ -69,6 +68,11 @@ class AMQPQueue extends Queue implements QueueContract
      */
     private $declareQueues;
 
+    /**
+     * @var int
+     */
+    private $retryAfter;
+
     /**
      * @param AMQPStreamConnection $connection
      * @param string               $defaultQueueName  Default queue name
@@ -83,6 +87,7 @@ class AMQPQueue extends Queue implements QueueContract
      * @param string               $exchangeName      Exchange name
      * @param mixed                $exchangeType      Exchange type
      * @param mixed                $exchangeFlags     Exchange flags
+     * @param mixed                $retryAfter       Optional timeout for failed jobs
      */
     public function __construct(
         AMQPStreamConnection $connection,
@@ -93,7 +98,8 @@ public function __construct(
         $defaultChannelId = null,
         $exchangeName = '',
         $exchangeType = null,
-        $exchangeFlags = []
+        $exchangeFlags = [],
+        $retryAfter = 0
     ) {
         $this->connection = $connection;
         $this->defaultQueueName = $defaultQueueName ?: 'default';
@@ -103,6 +109,7 @@ public function __construct(
         $this->defaultChannelId = $defaultChannelId;
         $this->exchangeName = $exchangeName;
         $this->channel = $connection->channel($this->defaultChannelId);
+        $this->retryAfter = $retryAfter;
 
         if ($exchangeName !== null) {
             $this->declareExchange($exchangeName, $exchangeType, $exchangeFlags);
@@ -135,22 +142,29 @@ protected function declareExchange($exchangeName, $exchangeType, array $exchange
         call_user_func_array([$this->channel, 'exchange_declare'], $flags);
     }
 
+    /**
+     * @return array
+     */
+    public function getCustomMessageOptions(){
+        return ['retryAfter' => $this->retryAfter];
+    }
+
     /**
      * Push a new job onto the queue.
      *
      * @param  string $job   Job implementation class name
      * @param  mixed  $data  Job custom data. Usually array
      * @param  string $queue Queue name, if different from the default one
      *
-     * @throws InvalidPayloadException
+     * @throws \Illuminate\Queue\InvalidPayloadException
      * @throws AMQPException
      * @return bool Always true
      */
     public function push($job, $data = '', $queue = null)
     {
         $queue = $this->prepareQueue($queue);
-        $amqpMessage = $this->prepareMessage($job, $data);
-        $this->channel->basic_publish($amqpMessage, $this->exchangeName, $this->getRoutingKey($queue));
+        $payload = new AMQPMessage($this->createPayload($job, $queue, $data), $this->messageProperties);
+        $this->channel->basic_publish($payload, $this->exchangeName, $this->getRoutingKey($queue));
 
         return true;
     }
@@ -163,14 +177,14 @@ public function push($job, $data = '', $queue = null)
      * @param string $queue Queue name, if different from the default one
      *
      * @return bool
-     * @throws InvalidPayloadException
+     * @throws \Illuminate\Queue\InvalidPayloadException
      * @throws AMQPException
      */
     public function addMessageToBatch($job, $data = '', $queue = null)
     {
         $queue = $this->prepareQueue($queue);
-        $amqpMessage = $this->prepareMessage($job, $data);
-        $this->channel->batch_basic_publish($amqpMessage, $this->exchangeName, $this->getRoutingKey($queue));
+        $payload = new AMQPMessage($this->createPayload($job, $queue, $data), $this->messageProperties);
+        $this->channel->batch_basic_publish($payload, $this->exchangeName, $this->getRoutingKey($queue));
 
         return true;
     }
@@ -270,9 +284,6 @@ protected function getRoutingKey($queue)
      */
     public function pushRaw($payload, $queue = null, array $options = [])
     {
-        // NB: DYNAMIC PRIORITY IS NOT IMPLEMENTED FOR RAW MESSAGES,
-        // NB: NEED TO SET THE $this->messageProperties['priority'] FIELD
-
         $queue = $this->prepareQueue($queue);
         $amqpPayload = new AMQPMessage($payload, $this->messageProperties);
         $this->channel->basic_publish($amqpPayload, $this->exchangeName, $queue);
@@ -288,7 +299,7 @@ public function pushRaw($payload, $queue = null, array $options = [])
      * @param  string        $queue Queue name, if different from the default one
      *
      * @return bool Always true
-     * @throws InvalidPayloadException
+     * @throws \Illuminate\Queue\InvalidPayloadException
      * @throws AMQPException
      */
     public function later($delay, $job, $data = '', $queue = null)
@@ -300,8 +311,8 @@ public function later($delay, $job, $data = '', $queue = null)
         $queue = $this->prepareQueue($queue);
         $delayedQueueName = $this->declareDelayedQueue($queue, $delay);
 
-        $amqpMessage = $this->prepareMessage($job, $data);
-        $this->channel->basic_publish($amqpMessage, $this->exchangeName, $delayedQueueName);
+        $payload = new AMQPMessage($this->createPayload($job, $queue, $data), $this->messageProperties);
+        $this->channel->basic_publish($payload, $this->exchangeName, $delayedQueueName);
         return true;
     }
 
@@ -334,7 +345,7 @@ public function declareDelayedQueue($destinationQueueName, $delay)
             'arguments' => new AMQPTable([
                 'x-dead-letter-exchange'    => '',
                 'x-dead-letter-routing-key' => $destinationQueueName,
-                'x-message-ttl'             => $delay * 1000,
+                'x-message-ttl'             => intval($delay * 1000),
             ]),
         ]);
 
@@ -414,73 +425,55 @@ private function prepareQueue($queue)
         return $queue;
     }
 
-
-
-    // FNX - ADD DYNAMIC PRIORITY TO MESSAGE
-
     /**
-     * @param string $job
-     * @param mixed  $data
+     * Create a payload string from the given job and data.
+     *
+     * @param  string  $job
+     * @param  string  $queue
+     * @param  mixed   $data
+     * @return string
      *
-     * @return AMQPMessage
-     * @throws InvalidPayloadException
+     * @throws \Illuminate\Queue\InvalidPayloadException
      */
-    protected function prepareMessage($job, $data)
+    protected function createPayload($job, $queue, $data = '')
     {
-        $payloadJson = $this->createPayload($job, $data);
-        $arrPayload = json_decode($payloadJson, true);
-
-        // OVERRIDE PRIORITY
-        // NB: IMPLEMENT QUEUE DEFAULT PRIORITY USING THE FIELD `message_properties['priority']` IN THE CONFIG
-        $props = $this->messageProperties;
-        if (!empty($arrPayload['priority']) && $arrPayload['priority'] > 0)
-            $props['priority'] = $arrPayload['priority'];
-
-        return new AMQPMessage($payloadJson, $props);
+        $data = is_array($data) ? array_merge($data, $this->getCustomMessageOptions()) : $this->getCustomMessageOptions();
+        $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()
+            );
+        }
 
+        return $payload;
     }
 
-    // FNX - OVERRIDE PAYLOAD GENERATION
     /**
      * 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 [
-            'data' => [
-                'command' => serialize(clone $job),
-                'commandName' => get_class($job),
-            ],
+        $payload = $this->withCreatePayloadHooks($queue, [
             'displayName' => $this->getDisplayName($job),
             'job' => 'Illuminate\Queue\CallQueuedHandler@call',
-            'maxTries' => empty($job->tries) ? null : $job->tries,
-            'timeout' => empty($job->timeout) ? null : $job->timeout,
+            'maxTries' => $job->tries ?? null,
+            'timeout' => $job->timeout ?? null,
             'timeoutAt' => $this->getJobExpiration($job),
-            'priority' => empty($job->priority) ? null : $job->priority,
-        ];
-    }
-    /**
-     * Create a typical, string based queue payload array.
-     *
-     * @param  string  	$job
-     * @param  mixed  	$payload
-     *
-     * @return array
-     */
-    protected function createStringPayload($job, $payload)
-    {
-        return [
-            'displayName' => is_string($job) ? explode('@', $job)[0] : null,
-            'job' => $job,
-            'maxTries' 	=> $payload['maxTries'],
-            'timeout' 	=> $payload['timeout'],
-            'data' 		=> $payload['data'],
-            // BUG-FIX FOR ATTEMPTS HERE (OTHERWISE THIS FUNCTION IS THE SAME AS THE BASE IMPLEMENTATION):
-            'attempts' 	=> $payload['attempts'],
-            'priority'  => empty($payload['priority']) ? null : $payload['priority'],
-        ];
+            'data' => [
+                'commandName' => $job,
+                'command' => $job,
+            ],
+        ]);
+
+        return array_merge($payload, [
+            'data' => array_merge([
+                'commandName' => get_class($job),
+                'command' => serialize(clone $job),
+            ],$this->getCustomMessageOptions()),
+        ]);
     }
 }