Merge pull request vyuldashev#502 from shuqingzai/dev-13.0.1

fix vyuldashev#449 vyuldashev#499 vyuldashev#501

Author: M-Porter

src/Consumer.php

@@ -88,7 +88,7 @@ public function daemon($connectionName, $queue, WorkerOptions $options)
         $this->channel->basic_qos(
             $this->prefetchSize,
             $this->prefetchCount,
-            null
+            false
         );
 
         $jobClass = $connection->getJobClass();
@@ -147,7 +147,7 @@ function (AMQPMessage $message) use ($connection, $options, $connectionName, $qu
             } catch (AMQPRuntimeException $exception) {
                 $this->exceptions->report($exception);
 
-                $this->kill(1);
+                $this->kill(self::EXIT_ERROR, $options);
             } catch (Exception|Throwable $exception) {
                 $this->exceptions->report($exception);
 
@@ -171,7 +171,7 @@ function (AMQPMessage $message) use ($connection, $options, $connectionName, $qu
             );
 
             if (! is_null($status)) {
-                return $this->stop($status);
+                return $this->stop($status, $options);
             }
 
             $this->currentJob = null;
@@ -195,14 +195,15 @@ protected function daemonShouldRun(WorkerOptions $options, $connectionName, $que
      * Stop listening and bail out of the script.
      *
      * @param  int  $status
+     * @param  WorkerOptions|null  $options
      * @return int
      */
-    public function stop($status = 0): int
+    public function stop($status = 0, $options = null)
     {
         // Tell the server you are going to stop consuming.
         // It will finish up the last message and not send you any more.
         $this->channel->basic_cancel($this->consumerTag, false, true);
 
-        return parent::stop($status);
+        return parent::stop($status, $options);
     }
 }

src/Horizon/RabbitMQQueue.php

@@ -64,7 +64,7 @@ public function later($delay, $job, $data = '', $queue = null)
     {
         $payload = (new JobPayload($this->createPayload($job, $data)))->prepare($job)->value;
 
-        return tap(parent::pushRaw($payload, $queue, ['delay' => $this->secondsUntil($delay)]), function () use ($payload, $queue): void {
+        return tap(parent::laterRaw($delay, $payload, $queue), function () use ($payload, $queue): void {
             $this->event($this->getQueue($queue), new JobPushed($payload));
         });
     }

src/Queue/RabbitMQQueue.php

@@ -188,11 +188,18 @@ public function laterRaw($delay, $payload, $queue = null, $attempts = 0)
     {
         $ttl = $this->secondsUntil($delay) * 1000;
 
+        // default options
+        $options = ['delay' => $delay, 'attempts' => $attempts];
+
         // When no ttl just publish a new message to the exchange or queue
         if ($ttl <= 0) {
-            return $this->pushRaw($payload, $queue, ['delay' => $delay, 'attempts' => $attempts]);
+            return $this->pushRaw($payload, $queue, $options);
         }
 
+        // Create a main queue to handle delayed messages
+        [$mainDestination, $exchange, $exchangeType, $attempts] = $this->publishProperties($queue, $options);
+        $this->declareDestination($mainDestination, $exchange, $exchangeType);
+
         $destination = $this->getQueue($queue).'.delay.'.$ttl;
 
         $this->declareQueue($destination, true, false, $this->getDelayQueueArguments($this->getQueue($queue), $ttl));