Fix local vyuldashev#501 - laravel 9.41

Author: shuqingzai

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));
@@ -754,9 +761,9 @@ protected function getRoutingKey(string $destination): string
     protected function getExchangeType(?string $type = null): string
     {
         return @constant(AMQPExchangeType::class.'::'.Str::upper($type ?: Arr::get(
-            $this->options,
-            'exchange_type'
-        ) ?: 'direct')) ?: AMQPExchangeType::DIRECT;
+                $this->options,
+                'exchange_type'
+            ) ?: 'direct')) ?: AMQPExchangeType::DIRECT;
     }
 
     /**