retry-after option related to #21

increment required php version to real value

Author: martianoff

README.md

@@ -41,6 +41,9 @@ 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,
         ),
 ),
 ```

composer.json

@@ -4,7 +4,7 @@
   "minimum-stability": "stable",
   "license": "GPL-2.0",
   "require": {
-    "php": ">=5.4.0",
+    "php": ">=7",
     "php-amqplib/php-amqplib": "2.6.*"
   },
   "require-dev": {

src/Connectors/AmqpConnector.php

@@ -50,7 +50,7 @@ public function connect(array $config)
         return new AMQPQueue(
             $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

@@ -75,6 +75,10 @@ public function release($delay = 0)
         $job = $body['job'];
         $data = $body['data'];
 
+        //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) {
@@ -127,4 +131,5 @@ public function getJobId()
     {
         return $this->amqpMessage->get('message_id');
     }
+
 }

src/Queue/AMQPQueue.php

@@ -6,6 +6,7 @@
 use Forumhouse\LaravelAmqp\Jobs\AMQPJob;
 use Forumhouse\LaravelAmqp\Utility\ArrayUtil;
 use Illuminate\Contracts\Queue\Queue as QueueContract;
+use Illuminate\Queue\InvalidPayloadException;
 use Illuminate\Queue\Queue;
 use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Connection\AMQPConnection;
@@ -68,6 +69,11 @@ class AMQPQueue extends Queue implements QueueContract
      */
     private $declareQueues;
 
+    /**
+     * @var int
+     */
+    private $retryAfter;
+
     /**
      * @param AMQPStreamConnection $connection
      * @param string               $defaultQueueName  Default queue name
@@ -82,6 +88,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,
@@ -92,7 +99,8 @@ public function __construct(
         $defaultChannelId = null,
         $exchangeName = '',
         $exchangeType = null,
-        $exchangeFlags = []
+        $exchangeFlags = [],
+        $retryAfter = 0
     ) {
         $this->connection = $connection;
         $this->defaultQueueName = $defaultQueueName ?: 'default';
@@ -102,6 +110,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);
@@ -134,6 +143,13 @@ 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.
      *
@@ -410,4 +426,47 @@ private function prepareQueue($queue)
 
         return $queue;
     }
+
+    /**
+     * Create a payload string from the given job and data.
+     *
+     * @param  string  $job
+     * @param  mixed   $data
+     * @return string
+     *
+     * @throws \Illuminate\Queue\InvalidPayloadException
+     */
+    protected function createPayload($job, $data = '')
+    {
+        $data = is_array($data) ? array_merge($data, $this->getCustomMessageOptions()) : $this->getCustomMessageOptions();
+        $payload = json_encode($this->createPayloadArray($job, $data));
+        if (JSON_ERROR_NONE !== json_last_error()) {
+            throw new InvalidPayloadException(
+                'Unable to JSON encode payload. Error code: '.json_last_error()
+            );
+        }
+
+        return $payload;
+    }
+
+    /**
+     * Create a payload for an object-based queue handler.
+     *
+     * @param  mixed  $job
+     * @return array
+     */
+    protected function createObjectPayload($job)
+    {
+        return [
+            'displayName' => $this->getDisplayName($job),
+            'job' => 'Illuminate\Queue\CallQueuedHandler@call',
+            'maxTries' => $job->tries ?? null,
+            'timeout' => $job->timeout ?? null,
+            'timeoutAt' => $this->getJobExpiration($job),
+            'data' => array_merge([
+                'commandName' => get_class($job),
+                'command' => serialize(clone $job),
+            ],$this->getCustomMessageOptions()),
+        ];
+    }
 }