Merge branch 'custom_worker_patch' into custom_worker_patch_6

# Conflicts:
#	composer.json
#	composer.lock

Author: hex7c0

src/Queue/RabbitMQQueue.php

@@ -2,6 +2,8 @@
 
 namespace VladimirYuldashev\LaravelQueueRabbitMQ\Queue;
 
+use Illuminate\Support\Facades\Config;
+use PhpAmqpLib\Exception\AMQPChannelClosedException;
 use RuntimeException;
 use Illuminate\Queue\Queue;
 use Illuminate\Support\Str;
@@ -13,6 +15,7 @@
 use Interop\Amqp\Impl\AmqpBind;
 use Illuminate\Contracts\Queue\Queue as QueueContract;
 use VladimirYuldashev\LaravelQueueRabbitMQ\Queue\Jobs\RabbitMQJob;
+use Throwable;
 
 class RabbitMQQueue extends Queue implements QueueContract
 {
@@ -57,13 +60,13 @@ public function size($queueName = null): int
     }
 
     /** {@inheritdoc} */
-    public function push($job, $data = '', $queue = null)
+    public function push($job, $data = '', $queue = null): ?string
     {
         return $this->pushRaw($this->createPayload($job, $queue, $data), $queue, []);
     }
 
     /** {@inheritdoc} */
-    public function pushRaw($payload, $queueName = null, array $options = [])
+    public function pushRaw($payload, $queueName = null, array $options = []): ?string
     {
         try {
             /**
@@ -128,12 +131,12 @@ public function pushRaw($payload, $queueName = null, array $options = [])
         } catch (\Exception $exception) {
             $this->reportConnectionError('pushRaw', $exception);
 
-            return;
+            return null;
         }
     }
 
     /** {@inheritdoc} */
-    public function later($delay, $job, $data = '', $queue = null)
+    public function later($delay, $job, $data = '', $queue = null): ?string
     {
         return $this->pushRaw($this->createPayload($job, $queue, $data), $queue, ['delay' => $this->secondsUntil($delay)]);
     }
@@ -206,18 +209,23 @@ public function getContext(): AmqpContext
     }
 
     /**
-     * @param string $queueName
+     * @param string|null $queueName
      *
      * @return array [Interop\Amqp\AmqpQueue, Interop\Amqp\AmqpTopic]
      */
-    protected function declareEverything(string $queueName = null): array
+    protected function declareEverything(?string $queueName = null): array
     {
+        /** @var string $queueName */
         $queueName = $this->getQueueName($queueName);
+
+        /** @var string $exchangeName */
         $exchangeName = $this->exchangeOptions['name'] ?: $queueName;
 
         $topic = $this->context->createTopic($exchangeName);
+
         $topic->setType($this->exchangeOptions['type']);
         $topic->setArguments($this->exchangeOptions['arguments']);
+
         if ($this->exchangeOptions['passive']) {
             $topic->addFlag(AmqpTopic::FLAG_PASSIVE);
         }
@@ -229,45 +237,61 @@ protected function declareEverything(string $queueName = null): array
         }
 
         if ($this->exchangeOptions['declare'] && ! in_array($exchangeName, $this->declaredExchanges, true)) {
-            $this->context->declareTopic($topic);
+            try {
+                $this->context->declareTopic($topic);
+            } catch (AMQPChannelClosedException $e) {
+                throw new AMQPChannelClosedException('Exchange declared with different arguments.', 0, $e);
+            }
 
             $this->declaredExchanges[] = $exchangeName;
         }
 
         $queue = $this->context->createQueue($queueName);
-        $queue->setArguments($this->queueOptions['arguments']);
-        if ($this->queueOptions['passive']) {
+
+        /** @var array $queueOptions */
+        $queueOptions = Config::get("queue.connections.rabbitmq.options");
+
+        $queueOptions = $queueOptions["queue_{$queueName}"] ?? $this->queueOptions;
+
+        if ($queueOptions['arguments']) {
+            $queue->setArguments(is_string($queueOptions['arguments']) ? json_decode($queueOptions['arguments'], true) : $queueOptions['arguments']);
+        }
+        if ($queueOptions['passive']) {
             $queue->addFlag(AmqpQueue::FLAG_PASSIVE);
         }
-        if ($this->queueOptions['durable']) {
+        if ($queueOptions['durable']) {
             $queue->addFlag(AmqpQueue::FLAG_DURABLE);
         }
-        if ($this->queueOptions['exclusive']) {
+        if ($queueOptions['exclusive']) {
             $queue->addFlag(AmqpQueue::FLAG_EXCLUSIVE);
         }
-        if ($this->queueOptions['auto_delete']) {
+        if ($queueOptions['auto_delete']) {
             $queue->addFlag(AmqpQueue::FLAG_AUTODELETE);
         }
 
-        if ($this->queueOptions['declare'] && ! in_array($queueName, $this->declaredQueues, true)) {
-            $this->context->declareQueue($queue);
+        if ($queueOptions['declare'] && ! in_array($queueName, $this->declaredQueues, true)) {
+            try {
+                $this->context->declareQueue($queue);
+            } catch (AMQPChannelClosedException $e) {
+                throw new AMQPChannelClosedException('Queue declared with different arguments.', 0, $e);
+            }
 
             $this->declaredQueues[] = $queueName;
         }
 
-        if ($this->queueOptions['bind']) {
+        if ($queueOptions['bind']) {
             $this->context->bind(new AmqpBind($queue, $topic, $queue->getQueueName()));
         }
 
         return [$queue, $topic];
     }
 
-    protected function getQueueName($queueName = null)
+    protected function getQueueName($queueName = null): string
     {
         return $queueName ?: $this->queueName;
     }
 
-    protected function createPayloadArray($job, $queue, $data = '')
+    protected function createPayloadArray($job, $queue, $data = ''): array
     {
         return array_merge(parent::createPayloadArray($job, $queue, $data), [
             'id' => $this->getRandomId(),
@@ -286,10 +310,11 @@ protected function getRandomId(): string
 
     /**
      * @param string $action
-     * @param \Throwable $e
-     * @throws \Exception
+     * @param Throwable $e
+     *
+     * @throws RuntimeException
      */
-    protected function reportConnectionError($action, \Throwable $e)
+    protected function reportConnectionError(string $action, Throwable $e): void
     {
         /** @var LoggerInterface $logger */
         $logger = $this->container['log'];