0.9 version

composer.json

@@ -5,17 +5,18 @@
     "keywords": ["messaging", "queue", "laravel"],
     "license": "MIT",
     "require": {
-        "php": ">=5.6",
-        "illuminate/queue": "^5.4",
-        "queue-interop/queue-interop": "^0.6",
-        "queue-interop/amqp-interop": "^0.7",
-        "enqueue/amqp-tools": "^0.8"
+        "php": ">=7.1",
+        "illuminate/queue": "^5.6",
+        "queue-interop/amqp-interop": "0.8.x-dev",
+        "queue-interop/queue-interop": "0.7.x-dev",
+        "enqueue/enqueue": "0.9.x-dev",
+        "enqueue/dsn": "0.9.x-dev"
     },
     "require-dev": {
         "phpunit/phpunit": "~5.5",
-        "enqueue/enqueue": "^0.8@dev",
-        "enqueue/null": "^0.8@dev",
-        "enqueue/test": "^0.8@dev"
+        "enqueue/enqueue": "0.9.x-dev",
+        "enqueue/null": "0.9.x-dev",
+        "enqueue/test": "0.9.x-dev"
     },
     "autoload": {
         "psr-4": { "Enqueue\\LaravelQueue\\": "src/" },
@@ -33,7 +34,7 @@
             ]
         },
         "branch-alias": {
-            "dev-master": "0.8.x-dev"
+            "dev-master": "0.9.x-dev"
         }
     }
 }

src/AmqpQueue.php

@@ -5,18 +5,18 @@
 use Interop\Amqp\AmqpContext;
 
 /**
- * @method AmqpContext getPsrContext()
+ * @method AmqpContext getQueueInteropContext()
  */
 class AmqpQueue extends Queue
 {
     /**
      * {@inheritdoc}
      *
-     * @param AmqpContext $psrContext
+     * @param AmqpContext $amqpContext
      */
-    public function __construct(AmqpContext $psrContext, $queueName, $timeToRun)
+    public function __construct(AmqpContext $amqpContext, $queueName, $timeToRun)
     {
-        parent::__construct($psrContext, $queueName, $timeToRun);
+        parent::__construct($amqpContext, $queueName, $timeToRun);
     }
 
     /**
@@ -54,9 +54,9 @@ public function pop($queue = null)
      */
     protected function declareQueue($queue = null)
     {
-        $psrQueue = $this->getQueue($queue);
-        $psrQueue->addFlag(\Interop\Amqp\AmqpQueue::FLAG_DURABLE);
+        $interopQueue = $this->getQueue($queue);
+        $interopQueue->addFlag(\Interop\Amqp\AmqpQueue::FLAG_DURABLE);
 
-        $this->getPsrContext()->declareQueue($psrQueue);
+        $this->getQueueInteropContext()->declareQueue($queue);
     }
 }

src/Command/ConsumeCommand.php

@@ -0,0 +1,19 @@
+<?php
+namespace Enqueue\LaravelQueue\Command;
+
+use Enqueue\Container\Container;
+use Enqueue\SimpleClient\SimpleClient;
+
+class ConsumeCommand extends \Enqueue\Symfony\Client\ConsumeCommand
+{
+    public function __construct(SimpleClient $client)
+    {
+        $container = new Container([
+            'queue_consumer' => $client->getQueueConsumer(),
+            'driver' => $client->getDriver(),
+            'processor' => $client->getDelegateProcessor()
+        ]);
+
+        parent::__construct($container, 'queue_consumer', 'driver', 'processor');
+    }
+}

src/Command/ProduceCommand.php

@@ -0,0 +1,17 @@
+<?php
+namespace Enqueue\LaravelQueue\Command;
+
+use Enqueue\Container\Container;
+use Enqueue\SimpleClient\SimpleClient;
+
+class ProduceCommand extends \Enqueue\Symfony\Client\ProduceCommand
+{
+    public function __construct(SimpleClient $client)
+    {
+        $container = new Container([
+            'producer' => $client->getProducer(),
+        ]);
+
+        parent::__construct($container, 'producer');
+    }
+}

src/Command/RoutesCommand.php

@@ -0,0 +1,17 @@
+<?php
+namespace Enqueue\LaravelQueue\Command;
+
+use Enqueue\Container\Container;
+use Enqueue\SimpleClient\SimpleClient;
+
+class RoutesCommand extends \Enqueue\Symfony\Client\RoutesCommand
+{
+    public function __construct(SimpleClient $client)
+    {
+        $container = new Container([
+            'driver' => $client->getDriver(),
+        ]);
+
+        parent::__construct($container, 'driver');
+    }
+}

src/Command/SetupBrokerCommand.php

@@ -2,12 +2,17 @@
 namespace Enqueue\LaravelQueue\Command;
 
 
+use Enqueue\Container\Container;
 use Enqueue\SimpleClient\SimpleClient;
 
 class SetupBrokerCommand extends \Enqueue\Symfony\Client\SetupBrokerCommand
 {
     public function __construct(SimpleClient $client)
     {
-        parent::__construct($client->getDriver());
+        $container = new Container([
+            'driver' => $client->getDriver(),
+        ]);
+
+        parent::__construct($container, 'driver');
     }
 }

src/Connector.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\PsrConnectionFactory;
+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(PsrConnectionFactory::class)) {
-            throw new \LogicException(sprintf('The "connection_factory_class" option must contain a class that implements "%s" but it is not', PsrConnectionFactory::class));
-        }
+        /** @var ConnectionFactoryFactoryInterface $factory */
+        $factory = new $connectionFactoryFactoryClass();
+        $connection = $factory->create($config);
+        $context = $connection->createContext();
 
-        /** @var PsrConnectionFactory $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);
     }
 }

src/EnqueueServiceProvider.php

@@ -2,28 +2,21 @@
 
 namespace Enqueue\LaravelQueue;
 
-use Enqueue\LaravelQueue\Command\ConsumeMessagesCommand;
-use Enqueue\LaravelQueue\Command\ProduceMessageCommand;
-use Enqueue\LaravelQueue\Command\QueuesCommand;
+use Enqueue\LaravelQueue\Command\ConsumeCommand;
+use Enqueue\LaravelQueue\Command\ProduceCommand;
+use Enqueue\LaravelQueue\Command\RoutesCommand;
 use Enqueue\LaravelQueue\Command\SetupBrokerCommand;
-use Enqueue\LaravelQueue\Command\TopicsCommand;
 use Enqueue\SimpleClient\SimpleClient;
 use Illuminate\Queue\QueueManager;
 use Illuminate\Support\ServiceProvider;
 
 class EnqueueServiceProvider extends ServiceProvider
 {
-    /**
-     * {@inheritdoc}
-     */
     public function boot()
     {
         $this->bootInteropQueueDriver();
     }
 
-    /**
-     * {@inheritdoc}
-     */
     public function register()
     {
         $this->registerClient();
@@ -49,10 +42,9 @@ private function registerClient()
         if ($this->app->runningInConsole()) {
             $this->commands([
                 SetupBrokerCommand::class,
-                ProduceMessageCommand::class,
-                QueuesCommand::class,
-                TopicsCommand::class,
-                ConsumeMessagesCommand::class,
+                ProduceCommand::class,
+                RoutesCommand::class,
+                ConsumeCommand::class,
             ]);
         }
     }