[amqp] Set delay strategy if rabbitmq scheme extension present.

Author: makasim

AmqpConnectionFactory.php

@@ -7,6 +7,7 @@
 use Enqueue\AmqpTools\ConnectionConfig;
 use Enqueue\AmqpTools\DelayStrategyAware;
 use Enqueue\AmqpTools\DelayStrategyAwareTrait;
+use Enqueue\AmqpTools\RabbitMqDlxDelayStrategy;
 use Interop\Amqp\AmqpConnectionFactory as InteropAmqpConnectionFactory;
 use Interop\Queue\PsrContext;
 use PhpAmqpLib\Connection\AbstractConnection;
@@ -48,6 +49,10 @@ public function __construct($config = 'amqp:')
             ->addDefaultOption('keepalive', false)
             ->parse()
         ;
+
+        if (in_array('rabbitmq', $this->config->getSchemeExtensions(), true)) {
+            $this->setDelayStrategy(new RabbitMqDlxDelayStrategy());
+        }
     }
 
     /**

Tests/AmqpConnectionFactoryTest.php

@@ -3,6 +3,7 @@
 namespace Enqueue\AmqpLib\Tests;
 
 use Enqueue\AmqpLib\AmqpConnectionFactory;
+use Enqueue\AmqpTools\RabbitMqDlxDelayStrategy;
 use Enqueue\Test\ClassExtensionTrait;
 use Interop\Queue\PsrConnectionFactory;
 use PHPUnit\Framework\TestCase;
@@ -16,14 +17,10 @@ public function testShouldImplementConnectionFactoryInterface()
         $this->assertClassImplements(PsrConnectionFactory::class, AmqpConnectionFactory::class);
     }
 
-    public function testShouldSupportAmqpLibScheme()
+    public function testShouldSetRabbitMqDlxDelayStrategyIfRabbitMqSchemeExtensionPresent()
     {
-        // no exception here
-        new AmqpConnectionFactory('amqp+lib:');
-        new AmqpConnectionFactory('amqps+lib:');
+        $factory = new AmqpConnectionFactory('amqp+rabbitmq:');
 
-        $this->expectException(\LogicException::class);
-        $this->expectExceptionMessage('The given DSN scheme "amqp+foo" is not supported. Could be one of "amqp", "amqps", "amqp+lib');
-        new AmqpConnectionFactory('amqp+foo:');
+        $this->assertAttributeInstanceOf(RabbitMqDlxDelayStrategy::class, 'delayStrategy', $factory);
     }
 }

Tests/Spec/AmqpSendToTopicAndReceiveFromQueueWithBasicGetMethodTest.php Tests/Spec/AmqpSendToTopicAndReceiveFromQueueTest.php

@@ -12,7 +12,7 @@
 /**
  * @group functional
  */
-class AmqpSendToTopicAndReceiveFromQueueWithBasicGetMethodTest extends SendToTopicAndReceiveFromQueueSpec
+class AmqpSendToTopicAndReceiveFromQueueTest extends SendToTopicAndReceiveFromQueueSpec
 {
     /**
      * {@inheritdoc}

Tests/Spec/AmqpSendToTopicAndReceiveFromQueueWithBasicConsumeMethodTest.php

@@ -10,7 +10,6 @@
     'port' => 5672,
     'user' => 'guest',
     'pass' => 'guest',
-    'receive_method' => 'basic_consume',
 ];
 
 $connection = new AmqpConnectionFactory($config);

tutorial/receive.php

@@ -12,7 +12,6 @@
     'port' => 5672,
     'user' => 'guest',
     'pass' => 'guest',
-    'receive_method' => 'basic_consume',
 ];
 
 $connection = new AmqpConnectionFactory($config);

tutorial/receive_logs.php

@@ -12,7 +12,6 @@
     'port' => 5672,
     'user' => 'guest',
     'pass' => 'guest',
-    'receive_method' => 'basic_consume',
 ];
 
 $connection = new AmqpConnectionFactory($config);

tutorial/receive_logs_direct.php

@@ -12,7 +12,6 @@
     'port' => 5672,
     'user' => 'guest',
     'pass' => 'guest',
-    'receive_method' => 'basic_consume',
 ];
 
 $connection = new AmqpConnectionFactory($config);

tutorial/receive_logs_topic.php

@@ -9,7 +9,6 @@
     'port' => 5672,
     'user' => 'guest',
     'pass' => 'guest',
-    'receive_method' => 'basic_consume',
 ];
 
 class FibonacciRpcClient

tutorial/rpc_client.php

@@ -9,7 +9,6 @@
     'port' => 5672,
     'user' => 'guest',
     'pass' => 'guest',
-    'receive_method' => 'basic_consume',
 ];
 
 function fib($n)

tutorial/rpc_server.php

@@ -10,7 +10,6 @@
     'port' => 5672,
     'user' => 'guest',
     'pass' => 'guest',
-    'receive_method' => 'basic_consume',
 ];
 
 $connection = new AmqpConnectionFactory($config);