Use connection factory factory.

Author: makasim

composer.json

@@ -9,7 +9,8 @@
         "illuminate/queue": "^5.6",
         "queue-interop/amqp-interop": "0.8.x-dev",
         "queue-interop/queue-interop": "0.7.x-dev",
-        "enqueue/amqp-tools": "0.9.x-dev"
+        "enqueue/enqueue": "0.9.x-dev",
+        "enqueue/dsn": "0.9.x-dev"
     },
     "require-dev": {
         "phpunit/phpunit": "~5.5",

src/AmqpConnector.php

@@ -2,39 +2,54 @@
 
 namespace Enqueue\LaravelQueue;
 
+use Enqueue\AmqpTools\DelayStrategy;
+use Enqueue\AmqpTools\DelayStrategyAware;
+use Enqueue\AmqpTools\RabbitMqDelayPluginDelayStrategy;
+use Enqueue\AmqpTools\RabbitMqDlxDelayStrategy;
+use Enqueue\ConnectionFactoryFactory;
+use Enqueue\ConnectionFactoryFactoryInterface;
+use Illuminate\Contracts\Queue\Queue as QueueContract;
 use Illuminate\Queue\Connectors\ConnectorInterface;
-use Interop\Queue\ConnectionFactory;
+use Interop\Amqp\AmqpContext;
 
 class Connector implements ConnectorInterface
 {
-    /**
-     * {@inheritdoc}
-     */
-    public function connect(array $config)
+    public function connect(array $config): QueueContract
     {
         $config = array_replace([
-            'connection_factory_class' => null,
+            'dsn' => null,
+            'factory_class' => null,
             'queue' => 'default',
             'time_to_run' => 0,
         ], $config);
 
-        if (empty($config['connection_factory_class'])) {
-            throw new \LogicException('The "connection_factory_class" option is required');
-        }
+        $queue = $config['queue'];
+        $timeToRum = $config['time_to_run'];
+        $connectionFactoryFactoryClass = $config['factory_class'] ?? ConnectionFactoryFactory::class;
 
-        $factoryClass = $config['connection_factory_class'];
-        if (false == class_exists($factoryClass)) {
-            throw new \LogicException(sprintf('The "connection_factory_class" option "%s" is not a class', $factoryClass));
-        }
+        unset($config['factory_class']);
 
-        $rc = new \ReflectionClass($factoryClass);
-        if (false == $rc->implementsInterface(ConnectionFactory::class)) {
-            throw new \LogicException(sprintf('The "connection_factory_class" option must contain a class that implements "%s" but it is not', ConnectionFactory::class));
-        }
+        /** @var ConnectionFactoryFactoryInterface $factory */
+        $factory = new $connectionFactoryFactoryClass();
+        $connection = $factory->create($config);
+        $context = $connection->createContext();
 
-        /** @var ConnectionFactory $factory */
-        $factory = new $factoryClass($config);
+        if ($context instanceof AmqpContext) {
+            $config = array_replace(['delay_strategy' => 'rabbitmq_dlx'], $config);
+
+            if ($context instanceof DelayStrategyAware && 'rabbitmq_dlx' == $config['delay_strategy']) {
+                $context->setDelayStrategy(new RabbitMqDlxDelayStrategy());
+            }
+            if ($context instanceof DelayStrategyAware && 'rabbitmq_delay_plugin' == $config['delay_strategy']) {
+                $context->setDelayStrategy(new RabbitMqDelayPluginDelayStrategy());
+            }
+            if ($context instanceof DelayStrategyAware && $config['delay_strategy'] instanceof DelayStrategy) {
+                $context->setDelayStrategy($config['delay_strategy']);
+            }
+
+            return new AmqpQueue($context, $queue, $timeToRum);
+        }
 
-        return new Queue($factory->createContext(), $config['queue'], $config['time_to_run']);
+        return new Queue($context, $queue, $timeToRum);
     }
 }