Skip to content

WIP refactor separating queue backend from batch management #46

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mmenozzi wants to merge 9 commits into
base: master
Choose a base branch
from
Open

WIP refactor separating queue backend from batch management #46

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
1d797a9
WIP refactor separating queue backend from batch management
mmenozzi Jan 7, 2026
743d1d0
Fixes the test errors
azambon Jan 8, 2026
1efdbdf
Removes the deprecated release_delay configuration parameter
azambon Jan 12, 2026
315fd01
Merge remote-tracking branch 'origin/master' into queue-manger-race-c…
azambon Jan 12, 2026
6a918e9
Removes a bunch of unused deprecated exception classes
azambon Jan 12, 2026
cd6d99b
Removes the deprecated pagination functions
azambon Jan 12, 2026
560dfca
Fix CS
azambon Jan 12, 2026
e3cf9f6
Updates the build to include PHP 8.5 and run on the latest Ubuntu ver…
azambon Jan 12, 2026
b041d2f
Adds PHP 8.5 to the allowed PHP versions in the composer.json file
azambon Jan 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .github/workflows/ci.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 0 additions & 1 deletion README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
2 changes: 1 addition & 1 deletion composer.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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": "*",
Expand Down
45 changes: 0 additions & 45 deletions src/Console/Pager/AsyncPager.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
20 changes: 0 additions & 20 deletions src/Console/Pager/LessThan1CurrentPageException.php
View file
Open in desktop

This file was deleted.

17 changes: 0 additions & 17 deletions src/Console/Pager/LessThan1MaxPerPageException.php
View file
Open in desktop

This file was deleted.

19 changes: 0 additions & 19 deletions src/Console/Pager/NotBooleanException.php
View file
Open in desktop

This file was deleted.

19 changes: 0 additions & 19 deletions src/Console/Pager/NotIntegerCurrentPageException.php
View file
Open in desktop

This file was deleted.

19 changes: 0 additions & 19 deletions src/Console/Pager/NotIntegerException.php
View file
Open in desktop

This file was deleted.

19 changes: 0 additions & 19 deletions src/Console/Pager/NotIntegerMaxPerPageException.php
View file
Open in desktop

This file was deleted.

20 changes: 0 additions & 20 deletions src/Console/Pager/NotValidCurrentPageException.php
View file
Open in desktop

This file was deleted.

20 changes: 0 additions & 20 deletions src/Console/Pager/NotValidMaxPerPageException.php
View file
Open in desktop

This file was deleted.

20 changes: 0 additions & 20 deletions src/Console/Pager/OutOfRangeCurrentPageException.php
View file
Open in desktop

This file was deleted.

1 change: 0 additions & 1 deletion src/FlowConfiguration.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
38 changes: 24 additions & 14 deletions src/FlowExtension.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
{
Expand Down Expand Up @@ -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) {
Expand All @@ -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);
Expand Down Expand Up @@ -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);
Expand Down
3 changes: 1 addition & 2 deletions src/Model/FlowConfig.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading
Loading