Enable dynamic queue configuration via the use of closures in queue_flags array

Author: FractalizeR

src/Queue/AMQPQueue.php

@@ -4,6 +4,7 @@
 
 use Forumhouse\LaravelAmqp\Exception\AMQPException;
 use Forumhouse\LaravelAmqp\Jobs\AMQPJob;
+use Forumhouse\LaravelAmqp\Utility\ArrayUtil;
 use Illuminate\Queue\Queue;
 use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Connection\AMQPConnection;
@@ -211,7 +212,7 @@ protected function declareExchange($exchangeName, $exchangeType, array $exchange
     public function declareQueue($name)
     {
         $queue = $this->getQueueName($name);
-        $flags = array_replace([
+        $flags = array_replace_recursive([
             'queue' => $queue,
             'passive' => false,
             'durable' => false,
@@ -220,7 +221,7 @@ public function declareQueue($name)
             'nowait' => false,
             'arguments' => null,
             'ticket' => null,
-        ], $this->queueFlags);
+        ], $this->getQueueFlags($name));
 
         call_user_func_array([$this->channel, 'queue_declare'], $flags);
     }
@@ -238,7 +239,7 @@ public function declareDelayedQueue($destinationQueueName, $delay)
         $destinationQueueName = $this->getQueueName($destinationQueueName);
         $deferredQueueName = $destinationQueueName . '_deferred_' . $delay;
 
-        $flags = array_replace([
+        $flags = array_replace_recursive([
             'queue' => '',
             'passive' => false,
             'durable' => false,
@@ -247,7 +248,7 @@ public function declareDelayedQueue($destinationQueueName, $delay)
             'nowait' => false,
             'arguments' => null,
             'ticket' => null,
-        ], $this->queueFlags, [
+        ], $this->getQueueFlags($destinationQueueName, $deferredQueueName, $delay), [
             'queue' => $deferredQueueName,
             'durable' => true,
             'arguments' => [
@@ -277,4 +278,22 @@ protected function getQueueName($name)
         }
         return $name;
     }
+
+    /**
+     * @param string      $queueName
+     * @param null|string $deferredQueueName
+     * @param null|int    $deferredQueueDelay
+     *
+     * @return array
+     */
+    protected function getQueueFlags($queueName, $deferredQueueName = null, $deferredQueueDelay = null)
+    {
+        $result = ArrayUtil::arrayMapRecursive(function ($value) {
+            return is_callable($value) ? call_user_func_array($value, func_get_args()) : $value;
+        }, $this->queueFlags);
+
+        $result = ArrayUtil::removeNullsRecursive($result);
+
+        return $result;
+    }
 }

src/Utility/ArrayUtil.php

@@ -0,0 +1,60 @@
+<?php
+
+namespace Forumhouse\LaravelAmqp\Utility;
+
+/**
+ * Various array utilities
+ *
+ * @package Forumhouse\LaravelAmqp\Utility
+ */
+class ArrayUtil
+{
+    /**
+     * array_map_recursive, which is missing from PHP, with the same signature
+     *
+     * @param callable $func
+     * @param array    $arr
+     *
+     * @return array
+     */
+    public static function arrayMapRecursive(callable $func, array $arr)
+    {
+        array_walk_recursive($arr, function (&$v) use ($func) {
+            $v = $func($v);
+        });
+        return $arr;
+    }
+
+    /**
+     * array_filter_recursive which is missing from PHP
+     *
+     * @param array    $input
+     * @param callable $callback
+     *
+     * @return array
+     */
+    public static function arrayFilterRecursive(array $input, callable $callback)
+    {
+        foreach ($input as &$value) {
+            if (is_array($value)) {
+                $value = static::arrayFilterRecursive($value, $callback);
+            }
+        }
+
+        return array_filter($input, $callback);
+    }
+
+    /**
+     * Returns array with all
+     *
+     * @param array $arr
+     *
+     * @return array
+     */
+    public static function removeNullsRecursive(array $arr)
+    {
+        return static::arrayFilterRecursive($arr, function ($var) {
+            return !is_null($var);
+        });
+    }
+}