Merge pull request #10 from lamoda/feature/SCENTRE-5889_extend_retry_sending_time

SCENTRE-5889 Add availability of select special strategy for queue

Author: dedpikhto

.gitignore

@@ -1,3 +1,3 @@
 /vendor/
 /composer.lock
-/.php_cs.cache
+/.php_cs.cache

README.md

@@ -37,7 +37,7 @@ Symfony bundle for convenient work with queues. Currently it supports RabbitMQ.
        max_attempts: 5
        batch_size_per_requeue: 5
        batch_size_per_republish: 5
-       ## optional
+       ## optional (will use for default delay Geometric Progression Strategy)
        strategy_delay_geometric_progression_start_interval_sec: 60
        strategy_delay_geometric_progression_multiplier: 2
    ```
@@ -199,6 +199,47 @@ Symfony bundle for convenient work with queues. Currently it supports RabbitMQ.
                 - { name: queue.handler, handle: App\Job\SendNotificationJob }
     ```
 
+1. Configure delay strategy parameters (optional)
+   
+   ```yaml
+   lamoda_queue:
+   ...
+       queues:
+           queue_one: 'delay_arithmetic_progression'
+           queue_two: 'delay_special_geometric_progression'
+   
+   # Settings of special behaviors for Delay strategies (optional)
+   services:
+     lamoda_queue.strategy.delay.arithmetic_progression:
+       class: Lamoda\QueueBundle\Strategy\Delay\ArithmeticProgressionStrategy
+       tags:
+         - { name: 'lamoda_queue_strategy', key: 'delay_arithmetic_progression' }
+       arguments:
+         - 60 # start_interval_sec parameter
+         - 1700 # multiplier parameter
+     
+    lamoda_queue.strategy.delay.geometric_progression:
+       class: Lamoda\QueueBundle\Strategy\Delay\GeometricProgressionStrategy
+       tags:
+         - { name: 'lamoda_queue_strategy', key: 'delay_special_geometric_progression' }
+       arguments:
+         - 70 # start_interval_sec parameter
+         - 4 # multiplier parameter
+
+   ```
+   In this block, you can config special delay behaviors for each queue. For this, you have to register new services that 
+   use one of several base strategies (ArithmeticProgressionStrategy, GeometricProgressionStrategy) or yours (for this you 
+   have to make Service Class that implements DelayStrategyInterface).
+   
+   Each strategy service has to have a tag with name `lamoda_queue_strategy` and unique `key`.
+   After this, you can use these `keys` for matching with queues in `lamoda_queue.queues` section.
+   
+   By default, use GeometricProgressionStrategy with params (their you can customize in `lamoda_queue` config section):
+   ```
+       strategy_delay_geometric_progression_start_interval_sec: 60
+       strategy_delay_geometric_progression_multiplier: 2
+   ```
+
 6. Add queue name in "codeception.yml" at `modules.config.AMQP.queues`
 
 7. Execute `./bin/console queue:init` command

composer.json

@@ -17,6 +17,7 @@
     "symfony/console": "^4.2 || ^5.0",
     "symfony/dependency-injection": "^4.2 || ^5.0",
     "symfony/event-dispatcher": "^4.3 || ^5.0",
+    "symfony/framework-bundle": "^4.3",
     "symfony/monolog-bundle": "^3.3"
   },
   "require-dev": {

src/DependencyInjection/Configuration.php

@@ -4,6 +4,7 @@
 
 namespace Lamoda\QueueBundle\DependencyInjection;
 
+use Symfony\Component\Config\Definition\Builder\ArrayNodeDefinition;
 use Symfony\Component\Config\Definition\Builder\TreeBuilder;
 use Symfony\Component\Config\Definition\ConfigurationInterface;
 
@@ -40,6 +41,22 @@ public function getConfigTreeBuilder(): TreeBuilder
                 ->end()
             ->end();
 
+        $this->addQueues($rootNode);
+
         return $treeBuilder;
     }
+
+
+    protected function addQueues(ArrayNodeDefinition $node)
+    {
+        $node
+            ->children()
+                ->arrayNode('queues')
+                ->canBeUnset()
+                ->useAttributeAsKey('key')
+                ->treatNullLike(array())
+                ->prototype('scalar')->end()
+            ->end()
+        ;
+    }
 }

src/DependencyInjection/LamodaQueueExtension.php

@@ -33,5 +33,6 @@ public function load(array $configs, ContainerBuilder $container): void
         $container->setParameter('lamoda_queue.strategy_delay_geometric_progression_start_interval_sec', $config['strategy_delay_geometric_progression_start_interval_sec']);
         $container->setParameter('lamoda_queue.strategy_delay_geometric_progression_multiplier', $config['strategy_delay_geometric_progression_multiplier']);
         $container->setParameter('lamoda_queue.command_unexpected_end_script_timeout', $config['command_unexpected_end_script_timeout']);
+        $container->setParameter('lamoda_queue.queues_configuration', $config['queues']);
     }
 }

src/Exception/UnknownStrategyKeyException.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Lamoda\QueueBundle\Exception;
+
+class UnknownStrategyKeyException extends \UnexpectedValueException
+{
+}

src/Resources/config/services.yml

@@ -65,13 +65,6 @@ services:
             - '@Lamoda\QueueBundle\Factory\EntityFactory'
             - '@event_dispatcher'
             - '%lamoda_queue.max_attempts%'
-    Lamoda\QueueBundle\Service\DelayService:
-        arguments:
-            - '@Lamoda\QueueBundle\Strategy\Delay\DelayStrategyInterface'
-    Lamoda\QueueBundle\Strategy\Delay\GeometricProgressionStrategy:
-        arguments:
-            - '%lamoda_queue.strategy_delay_geometric_progression_start_interval_sec%'
-            - '%lamoda_queue.strategy_delay_geometric_progression_multiplier%'
     Lamoda\QueueBundle\Command\QueueRepublishCommand:
         public: true
         arguments:
@@ -98,3 +91,17 @@ services:
         public: true
         tags:
             - { name: console.command }
+    Lamoda\QueueBundle\Service\DelayService:
+      arguments:
+        - '@Lamoda\QueueBundle\Service\DelayStrategyResolver'
+        - '@Psr\Log\LoggerInterface'
+    Lamoda\QueueBundle\Strategy\Delay\GeometricProgressionStrategy:
+      arguments:
+        - '%lamoda_queue.strategy_delay_geometric_progression_start_interval_sec%'
+        - '%lamoda_queue.strategy_delay_geometric_progression_multiplier%'
+      tags:
+        - { name: 'lamoda_queue_strategy', key: !php/const Lamoda\QueueBundle\Service\DelayStrategyResolver::DEFAULT_STRATEGY }
+    Lamoda\QueueBundle\Service\DelayStrategyResolver:
+        arguments:
+            - !tagged { tag: 'lamoda_queue_strategy', index_by: 'key' }
+            - '%lamoda_queue.queues_configuration%'

src/Service/DelayService.php

@@ -6,25 +6,43 @@
 
 use DateTime;
 use Lamoda\QueueBundle\Entity\QueueEntityInterface;
-use Lamoda\QueueBundle\Strategy\Delay\DelayStrategyInterface;
+use Lamoda\QueueBundle\Exception\UnknownStrategyKeyException;
+use Psr\Log\LoggerInterface;
 
 class DelayService
 {
     /**
-     * @var DelayStrategyInterface
+     * @var DelayStrategyResolver
      */
-    protected $strategy;
+    protected $strategyService;
 
-    public function __construct(DelayStrategyInterface $strategy)
+    /**
+     * @var LoggerInterface
+     */
+    private $logger;
+
+    public function __construct(DelayStrategyResolver $strategyService, LoggerInterface $logger)
     {
-        $this->strategy = $strategy;
+        $this->strategyService = $strategyService;
+        $this->logger          = $logger;
     }
 
     public function delayQueue(QueueEntityInterface $queue): QueueEntityInterface
     {
-        $iteration = $queue->getAttempts() ?? 1;
-        $newDelayInterval = $this->strategy->generateInterval($iteration);
-        $delayUntil = $this->getStartDateTime()->add($newDelayInterval);
+        try {
+            $strategy = $this->strategyService->getStrategy($queue->getName());
+        } catch (UnknownStrategyKeyException $exception) {
+            $this->logger->warning(
+                $exception->getMessage(), [
+                    'queue_name' => $queue->getName(),
+                ]
+            );
+            $strategy = $this->strategyService->getDefaultStrategy();
+        }
+
+        $iteration        = $queue->getAttempts() ?? 1;
+        $newDelayInterval = $strategy->generateInterval($iteration);
+        $delayUntil       = $this->getStartDateTime()->add($newDelayInterval);
 
         $queue->setWaiting($delayUntil);
 

src/Service/DelayStrategyResolver.php

@@ -0,0 +1,50 @@
+<?php
+declare(strict_types=1);
+
+namespace Lamoda\QueueBundle\Service;
+
+use Lamoda\QueueBundle\Exception\UnknownStrategyKeyException;
+use Lamoda\QueueBundle\Strategy\Delay\DelayStrategyInterface;
+
+class DelayStrategyResolver
+{
+    public const DEFAULT_STRATEGY = 'default_delay_strategy';
+
+    /**
+     * @var array
+     */
+    protected $handlers;
+
+    /**
+     * @var array
+     */
+    protected $queuesConfiguration;
+
+    public function __construct(iterable $handlers, array $queuesConfiguration = [])
+    {
+        $this->handlers            = iterator_to_array($handlers);
+        $this->queuesConfiguration = $queuesConfiguration;
+    }
+
+    /**
+     * @param string $queueName
+     *
+     * @return DelayStrategyInterface
+     * @throws UnknownStrategyKeyException
+     */
+    public function getStrategy(string $queueName): DelayStrategyInterface
+    {
+        $strategyKey = $this->queuesConfiguration[$queueName] ?? self::DEFAULT_STRATEGY;
+
+        if (isset($this->handlers[$strategyKey])) {
+            return $this->handlers[$strategyKey];
+        }
+
+        throw new UnknownStrategyKeyException(sprintf('Delay strategy with key: %s doesn\'t exist', $strategyKey));
+    }
+
+    public function getDefaultStrategy(): DelayStrategyInterface
+    {
+        return $this->handlers[self::DEFAULT_STRATEGY];
+    }
+}

src/Strategy/Delay/ArithmeticProgressionStrategy.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Lamoda\QueueBundle\Strategy\Delay;
+
+use DateInterval;
+
+class ArithmeticProgressionStrategy implements DelayStrategyInterface
+{
+    /** @var int */
+    private $startInterval;
+
+    /** @var float */
+    private $multiplier;
+
+    public function __construct(int $startIntervalSec, float $multiplier)
+    {
+        $this->startInterval = $startIntervalSec;
+        $this->multiplier    = $multiplier;
+    }
+
+    public function generateInterval(int $iteration): DateInterval
+    {
+        $newIntervalSec = (int) ceil($this->startInterval + ($this->multiplier * ($iteration - 1)));
+
+        return new DateInterval('PT' . $newIntervalSec . 'S');
+    }
+}