Refactor separating queue backend from batch management (ESB v3)

.github/workflows/ci.yaml

@@ -12,12 +12,12 @@ jobs:
   tests:
     name: "PHP ${{ matrix.php }}"
 
-    runs-on: ubuntu-22.04
+    runs-on: ubuntu-24.04
 
     strategy:
       fail-fast: false
       matrix:
-        php: [8.1, 8.2, 8.3, 8.4]
+        php: [8.1, 8.2, 8.3, 8.4, 8.5]
 
     env:
       ESB_CONSOLE_PORT: 8080

Dockerfile

@@ -1,4 +1,4 @@
-ARG PHP_VERSION=8.4
+ARG PHP_VERSION=8.5
 
 FROM php:${PHP_VERSION}-cli-alpine
 

README.md

@@ -28,7 +28,7 @@ Webgriffe ESB is designed to use a single binary which is used as the main entry
 
 Requirements
 ------------
-* PHP 8.1, 8.2, 8.3, 8.4
+* PHP 8.1, 8.2, 8.3, 8.4, 8.5
 * Beanstalk
 * Elasticsearch 7.*
 
@@ -72,7 +72,6 @@ flows:
       service: My\Esb\Worker            # A worker service ID defined above
       instances: 1                      # The number of worker instances to spawn for this flow
       error_retry_delay: 0              # The number of seconds to wait before an errored job can be retried. The default is 0 (errored jobs can be retried immediately). Useful when "retrying later" might solve the problem.
-      release_delay: 0                  # (deprecated) older name of the error_retry_delay parameter
       max_retry: 5                      # The number of maximum work retries for a job in this tube/flow before being buried
     dependencies:                       # This whole section can be omitted if the current flow has no dependencies
       flows: ['other_flow_1', 'other_flow_2']  # Optional: dependencies of this flow toward other flow(s)

composer.json

@@ -10,7 +10,7 @@
         }
     ],
     "require": {
-        "php": "~8.1.0|~8.2.0|~8.3.0|~8.4.0",
+        "php": "~8.1.0|~8.2.0|~8.3.0|~8.4.0|~8.5.0",
         "ext-pcntl": "*",
         "ext-json": "*",
         "ext-mbstring": "*",

src/Console/Pager/AsyncPager.php

@@ -224,57 +224,12 @@ public function getMaxPerPage()
      */
     public function setCurrentPage($currentPage)
     {
-        $this->useDeprecatedCurrentPageBooleanArguments(func_get_args());
-
         $this->currentPage = $this->filterCurrentPage($currentPage);
         $this->resetForCurrentPageChange();
 
         return $this;
     }
 
-    /**
-     * @param array<bool> $arguments
-     */
-    private function useDeprecatedCurrentPageBooleanArguments($arguments): void
-    {
-        $this->useDeprecatedCurrentPageAllowOutOfRangePagesBooleanArgument($arguments);
-        $this->useDeprecatedCurrentPageNormalizeOutOfRangePagesBooleanArgument($arguments);
-    }
-
-    /**
-     * @param array<bool> $arguments
-     */
-    private function useDeprecatedCurrentPageAllowOutOfRangePagesBooleanArgument($arguments): void
-    {
-        $index = 1;
-        $method = 'setAllowOutOfRangePages';
-
-        $this->useDeprecatedBooleanArgument($arguments, $index, $method);
-    }
-
-    /**
-     * @param array<bool> $arguments
-     */
-    private function useDeprecatedCurrentPageNormalizeOutOfRangePagesBooleanArgument($arguments): void
-    {
-        $index = 2;
-        $method = 'setNormalizeOutOfRangePages';
-
-        $this->useDeprecatedBooleanArgument($arguments, $index, $method);
-    }
-
-    /**
-     * @param array<bool> $arguments
-     * @param int $index
-     * @param string $method
-     */
-    private function useDeprecatedBooleanArgument($arguments, $index, $method): void
-    {
-        if (isset($arguments[$index])) {
-            $this->$method($arguments[$index]);
-        }
-    }
-
     /**
      * @param mixed $currentPage
      * @return int

src/FlowConfiguration.php

@@ -38,7 +38,6 @@ public function getConfigTreeBuilder(): TreeBuilder
                             ->scalarNode('service')->isRequired()->end()
                             ->integerNode('instances')->min(1)->defaultValue(1)->end()
                             ->integerNode('error_retry_delay')->min(0)->defaultValue(0)->end()
-                            ->integerNode('release_delay')->min(0)->defaultValue(0)->setDeprecated('Use "error_retry_delay" instead')->end()
                             ->integerNode('max_retry')->min(1)->defaultValue(5)->end()
                         ->end()
                     ->end()

src/FlowExtension.php

@@ -13,7 +13,9 @@
 use Symfony\Component\DependencyInjection\Extension\ExtensionInterface;
 use Symfony\Component\DependencyInjection\Reference;
 use Webgriffe\Esb\Model\FlowConfig;
-use Webgriffe\Esb\Service\QueueManager;
+use Webgriffe\Esb\Service\BatchManager;
+use Webgriffe\Esb\Service\BatchManagerFactory;
+use Webgriffe\Esb\Service\BeanstalkElasticsearchQueueBackend;
 
 final class FlowExtension implements ExtensionInterface, CompilerPassInterface
 {
@@ -66,10 +68,12 @@ public function process(ContainerBuilder $container): void
     {
         //These classes are defined manually. Remove the default definitions otherwise the container generates errors
         //trying to autowire them
-        $container->removeDefinition(QueueManager::class);
+        $container->removeDefinition(BeanstalkElasticsearchQueueBackend::class);
         $container->removeDefinition(FlowConfig::class);
         $container->removeDefinition(ProducerInstance::class);
         $container->removeDefinition(WorkerInstance::class);
+        $container->removeDefinition(BatchManager::class);
+        $container->removeDefinition(BatchManagerFactory::class);
 
         $flowManagerDefinition = $container->findDefinition(FlowManager::class);
         foreach ($this->flowsConfig as $flowName => $flowConfigData) {
@@ -78,30 +82,38 @@ public function process(ContainerBuilder $container): void
             $flowDefinition = new Definition(Flow::class);
             $flowDefinition->setAutowired(true);
             $flowDefinition->setArgument('$flowConfig', $flowConfig);
-            $queueManagerId = 'flow.queue_manager.' . $flowName;
+            $queueBackendId = 'flow.queue_backend.' . $flowName;
+            $batchManagerFactoryId = 'flow.batch_manager_factory.' . $flowName;
             try {
                 $producerDefinition = $container->findDefinition($flowConfig->getProducerServiceId());
                 $producerDefinition->setShared(false);
 
-                $queueManagerDefinition = new Definition();
-                $queueManagerDefinition
+                $queueBackendDefinition = new Definition();
+                $queueBackendDefinition
                     ->setShared(false)
                     ->setAutowired(true)
-                    ->setClass(QueueManager::class)
+                    ->setClass(BeanstalkElasticsearchQueueBackend::class)
                     ->setArgument('$flowConfig', $flowConfig)
+                ;
+                $container->setDefinition($queueBackendId, $queueBackendDefinition);
+
+                $batchManagerFactoryDefinition = new Definition();
+                $batchManagerFactoryDefinition
+                    ->setShared(false)
+                    ->setAutowired(true)
+                    ->setClass(BatchManagerFactory::class)
                     ->setArgument('$batchSize', $flowConfig->getProducerBatchSize())
                 ;
-                $container->setDefinition($queueManagerId, $queueManagerDefinition);
+                $container->setDefinition($batchManagerFactoryId, $batchManagerFactoryDefinition);
 
                 $producerInstanceDefinition = new Definition();
                 $producerInstanceDefinition
                     ->setAutowired(true)
                     ->setClass(ProducerInstance::class)
-                    ->setArgument('$producer', new Reference($flowConfig->getProducerServiceId()))
                     ->setArgument('$flowConfig', $flowConfig)
-                    ->setArgument('$queueManager', new Reference($queueManagerId))
-                    ->setArgument('$beanstalkClient', null)
-                    ->setArgument('$elasticSearch', null)
+                    ->setArgument('$producer', new Reference($flowConfig->getProducerServiceId()))
+                    ->setArgument('$queueBackend', new Reference($queueBackendId))
+                    ->setArgument('$batchManagerFactory', new Reference($batchManagerFactoryId))
                 ;
                 $producerInstanceId = 'flow.producer_instance' . $flowName;
                 $container->setDefinition($producerInstanceId, $producerInstanceDefinition);
@@ -129,9 +141,7 @@ public function process(ContainerBuilder $container): void
                         ->setArgument('$flowConfig', $flowConfig)
                         ->setArgument('$instanceId', $instanceId)
                         ->setArgument('$worker', new Reference($flowConfig->getWorkerServiceId()))
-                        ->setArgument('$queueManager', new Reference($queueManagerId))
-                        ->setArgument('$beanstalkClient', null)
-                        ->setArgument('$elasticSearch', null)
+                        ->setArgument('$queueBackend', new Reference($queueBackendId))
                     ;
                     $workerInstanceId = sprintf('flow.worker_instance.%s.%s', $flowName, $instanceId);
                     $container->setDefinition($workerInstanceId, $workerInstanceDefinition);

src/Model/FlowConfig.php

@@ -61,8 +61,7 @@ public function getWorkerInstancesCount(): int
 
     public function getWorkerReleaseDelay(): int
     {
-        //error_retry_delay is the new name, so give precedence to that
-        return $this->config['worker']['error_retry_delay'] ?: $this->config['worker']['release_delay'];
+        return $this->config['worker']['error_retry_delay'];
     }
 
     public function getWorkerMaxRetry(): int