fix publish job

Author: iamfarhad

src/Consumer.php

@@ -63,21 +63,17 @@ public function setPrefetchCount(int $value): void
      *
      * @throws Throwable
      */
-    public function daemon($connectionName, $queue, WorkerOptions $options): int
+    public function daemon($connectionName, $queue, WorkerOptions $options)
     {
         if ($this->supportsAsyncSignals()) {
             $this->listenForSignals();
         }
 
-        $config = (array) $options;
-
         $lastRestart = $this->getTimestampOfLastQueueRestart();
 
         [$startTime, $jobsProcessed] = [hrtime(true) / 1e9, 0];
 
         $connection = $this->manager->connection($connectionName);
-        $connection->declareQueue($queue);
-        $connection->setOptions($config);
 
         $this->channel = $connection->getChannel();
 
@@ -144,7 +140,7 @@ function (AMQPMessage $message) use ($connection, $options, $connectionName, $qu
                 $this->exceptions->report($exception);
 
                 $this->kill(1);
-            } catch (Exception | Throwable $exception) {
+            } catch (Exception|Throwable $exception) {
                 $this->exceptions->report($exception);
 
                 $this->stopWorkerIfLostConnection($exception);

src/RabbitQueue.php

@@ -63,8 +63,8 @@ public function push($job, $data = '', $queue = null)
             $this->createPayload($job, $this->getQueue($queue), $data),
             $queue,
             null,
-            function ($payload, $queue) use ($data) {
-                return $this->pushRaw(payload: $payload, queue: $queue);
+            function ($payload, $queue) {
+                return $this->pushRaw($payload, $queue);
             }
         );
     }
@@ -382,7 +382,7 @@ protected function createMessage($payload, int $attempts = 2): array
             'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
         ];
 
-        $currentPayload = json_decode($payload, true, 512, JSON_THROW_ON_ERROR);
+        $currentPayload = json_decode($payload, true, 512);
         if ($correlationId = $currentPayload['id'] ?? null) {
             $properties['correlation_id'] = $correlationId;
         }
@@ -415,13 +415,10 @@ protected function createMessage($payload, int $attempts = 2): array
     protected function publishProperties($queue, array $options = []): array
     {
         $queue = $this->getQueue($queue);
-        // @todo move to config
-        $attempts = 3;
+        $attempts = Arr::get($options, 'attempts') ?: 0;
 
         $destination = $queue;
         $exchange = $queue;
-
-        // @todo move to config
         $exchangeType = AMQPExchangeType::TOPIC;
 
         return [$destination, $exchange, $exchangeType, $attempts];