From 254a0d6a962199c504574c692396a7e3769adbc4 Mon Sep 17 00:00:00 2001 From: Nicolas Joubert Date: Tue, 20 Oct 2026 12:17:28 +0200 Subject: [PATCH] #307 - InputAggregatorTask: drop the items skipped by a branch, remove the deprecation - InputAggregatorTask no longer joins the inputs of two items when a branch skips one (filter, error with the skip strategy): the inputs received from the other branches are dropped when a common ancestor has been executed again since they were received (keep_inputs excepted). Inputs without a common ancestor are still joined. - New ProcessState::getExecutionCount(), incremented by the ProcessManager before each execution of a task (execute, proceed, flush). - Remove the deprecation of InputAggregatorTask (2018, notice added by #305, not released): the limits of the blocking tasks it referred to no longer exist, and a blocking task is not a replacement. Marked @final since 5.4 like the other concrete tasks. - Tests: unit tests of the task (first / last branch skipping, keep_inputs, no common ancestor), functional tests running the processes (filter and error with the skip strategy). Documentation and CHANGELOG updated. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 3 +- docs/05-good_practices.md | 3 +- docs/reference/tasks/input_aggregator_task.md | 48 +++++- src/Manager/ProcessManager.php | 1 + src/Model/ProcessState.php | 22 +++ src/Task/InputAggregatorTask.php | 70 +++++++-- tests/App/TestKernel.php | 45 ++++++ tests/Functional/KernelTest.php | 32 ++-- tests/Task/InputAggregatorTaskTest.php | 147 +++++++++++++----- 9 files changed, 307 insertions(+), 64 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 808d72eb..89a24625 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ Latest ------ ## Changes -* [#305](https://github.com/cleverage/process-bundle/issues/305) Bundle best practices, residues: the generic transformers are no longer autowired (explicit arguments). `ProcessHelpCommand` receives the `TaskRegistry` instead of the service container (passing a container is deprecated). Fetching from the service container a task of the application made public by the `TaskInterface` autoconfiguration is deprecated (`container.private` tag, new internal `PublicTaskDeprecationCompilerPass`); the tasks made public by the application itself are not concerned. Trigger the missing deprecation notices: `InputAggregatorTask`, `CheckSerializerCompilerPass`, `errors` task configuration key (use `error_outputs`). `TaskConfiguration::getErrors()` triggers a single notice on PHP 8.4+. `CleverAgeProcessBundle` is marked `@final since 5.4`. Update documentation, add tests. +* [#305](https://github.com/cleverage/process-bundle/issues/305) Bundle best practices, residues: the generic transformers are no longer autowired (explicit arguments). `ProcessHelpCommand` receives the `TaskRegistry` instead of the service container (passing a container is deprecated). Fetching from the service container a task of the application made public by the `TaskInterface` autoconfiguration is deprecated (`container.private` tag, new internal `PublicTaskDeprecationCompilerPass`); the tasks made public by the application itself are not concerned. Trigger the missing deprecation notices: `CheckSerializerCompilerPass`, `errors` task configuration key (use `error_outputs`). `TaskConfiguration::getErrors()` triggers a single notice on PHP 8.4+. `CleverAgeProcessBundle` is marked `@final since 5.4`. Update documentation, add tests. * [#299](https://github.com/cleverage/process-bundle/issues/299) Require `symfony/polyfill-php83`, `symfony/polyfill-php84` and `symfony/polyfill-php85`: the functions of PHP 8.3 to 8.5 can be used while supporting PHP 8.2. Rector targets PHP 8.2 (`withPhpVersion()`, `withPhpSets()` read from `composer.json`) and applies the rules using a newer function when its polyfill is required: `ProcessConfiguration` uses `array_any()` again; `array_first()` replaces `reset()` / `current()` in `ProcessConfiguration::getMainTask()`, `ConstraintLoader` and `ProcessHelpCommand` (no internal array pointer, same results). * [#279](https://github.com/cleverage/process-bundle/issues/279) Define the public API (Symfony BC promise): the classes not meant to be extended are marked `@final since 5.4` (the engine: `ProcessManager`, `ProcessConfigurationRegistry`, `TransformerRegistry`, `ContextualOptionResolver`; the models: `ProcessState`, `ProcessHistory`, `SubprocessInstance`, the configurations, the events, the `Filesystem` files; the concrete tasks and transformers), they will be `final` in v6.0: extending them triggers a deprecation with the Symfony `DebugClassLoader`. The abstract classes, interfaces, traits, exceptions and the file tasks extended by other tasks (`FileReaderTask`, `LineReaderTask`, `FolderBrowserTask`, `CsvReaderTask`, `InputCsvReaderTask`, `Filesystem\CsvResource`) stay open. `TaskRegistry`, new in this version, is `final`. The infrastructure classes are marked `@internal`: `CleverAgeProcessExtension`, `Configuration`, the commands, `DataQueueEventListener`, `PhpFunctionProvider`, the Monolog processors, `ProcessLogger` and `TaskLogger`. Update documentation. * [#269](https://github.com/cleverage/process-bundle/issues/269) Composer: `symfony/process`, `symfony/stopwatch`, `symfony/var-dumper` and `symfony/monolog-bundle` are no longer required, they are suggested (applications using them must require them). Without `symfony/process` (`CommandRunnerTask`, `ProcessLauncherTask`) or `symfony/stopwatch` (`StopwatchTask`), the container still compiles and these tasks fail with an explicit message (`UnavailableServicesCompilerPass`); without `symfony/var-dumper`, nothing is dumped (as before); without `symfony/monolog-bundle`, the `cleverage_process*` channels and processors are not registered and the default logger is used. `monolog/monolog` stays required (`Logger\*Processor`). Remove the unused `symfony/monolog-bridge` dependency. `ProcessConfiguration` no longer uses `array_any()` (PHP 8.4), whose polyfill was only installed by the removed packages. Add a `conflict` with `cleverage/doctrine-process-bundle`, `cleverage/soap-process-bundle` and `cleverage/ui-process-bundle` `<3.3`, which use the MonologBundle without requiring it. Update documentation, add tests. @@ -20,6 +20,7 @@ Latest * [#268](https://github.com/cleverage/process-bundle/issues/268) Clean up composer.json: remove the unused `symfony/dotenv`, `symfony/runtime` and `symfony/error-handler` dependencies (applications relying on the bundle to install them must require them), bound `symfony/service-contracts` to `^2.5|^3`, update the package description. ## Fixes +* [#307](https://github.com/cleverage/process-bundle/issues/307) Fix InputAggregatorTask: when a branch skips an item (filter, error with the `skip` strategy), the inputs received from the other branches for this item are dropped instead of being joined with the inputs of the next item. The inputs of an item are told apart by the executions of their common ancestors (new `ProcessState::getExecutionCount()`, incremented by the `ProcessManager` before each execution of a task). InputAggregatorTask is no longer deprecated (its limits described in 2018 no longer exist) and is marked `@final since 5.4`. Update documentation, add tests. * [#301](https://github.com/cleverage/process-bundle/issues/301) Fix ProcessManager: the history of the parent process is restored after a failed process too, so that the logs of a parent process going on after a failed sub-process (e.g. `process_executor` task with the `skip` error strategy) are no longer attributed to the sub-process (`process_id`, `process_code` and `process_context` added by the Monolog processors), and that no history of a failed process is left after its execution (`getProcessHistory()`). The `Critical process failure` log and the `cleverage_process.fail` event are still attributed to the failed process. `ProcessManager` implements `ResetInterface` (`kernel.reset` tag): its state is reset between two messages of a long-running worker. Add tests. v5.3 diff --git a/docs/05-good_practices.md b/docs/05-good_practices.md index fadcd204..2f23df41 100644 --- a/docs/05-good_practices.md +++ b/docs/05-good_practices.md @@ -71,7 +71,8 @@ See [errors and skips](04-advanced_workflow.md#errors-and-skips). When a task has several `outputs`, each output is fully processed one after the other, in the configured order, but this is an implementation detail. If a task needs the result of another branch, make the dependency explicit in the -graph (e.g. with a blocking task) rather than relying on the order in which branches are executed. +graph (e.g. with a blocking task or an [InputAggregatorTask](reference/tasks/input_aggregator_task.md)) rather than +relying on the order in which branches are executed. ## Use the context for runtime parameters diff --git a/docs/reference/tasks/input_aggregator_task.md b/docs/reference/tasks/input_aggregator_task.md index f6b136b7..b0fc1cfd 100644 --- a/docs/reference/tasks/input_aggregator_task.md +++ b/docs/reference/tasks/input_aggregator_task.md @@ -1,13 +1,13 @@ InputAggregatorTask =================== -**Deprecated since v5.4, removed in v6.0**: this task is too error-prone (it should be refactored as a blocking task). -An `E_USER_DEPRECATED` error is triggered when it is instantiated. - -Accumulates inputs coming from several parent tasks. The task is skipped until an input has been received from every +Joins the inputs coming from several parent tasks. The task is skipped until an input has been received from every parent declared in `input_codes`; it then outputs an array of all received inputs, indexed by their destination key, and clears its buffer (except for the keys listed in `keep_inputs`). +Unlike a blocking task (e.g. [AggregateIterableTask](aggregate_iterable_task.md)), which waits for all the items and +outputs them once, it outputs once **per item** in an iterable process. + Task reference -------------- @@ -62,10 +62,42 @@ aggregate: branch_b: b ``` +* Join two branches for each item of an iterable process, a branch skipping some items + +```yaml +# Task configuration level +entry: + task: constant_iterable_output + options: + output: [1, 2, 3] + outputs: [branch_a, branch_b] +branch_a: + task: filter + options: + not_match: { '': 2 } + outputs: [aggregate] +branch_b: + task: transformer + options: + transformers: + sprintf: { format: 'item %s' } + outputs: [aggregate] +aggregate: + task: input_aggregator + options: + input_codes: + branch_a: a + branch_b: b +# Outputs ['a' => 1, 'b' => 'item 1'], then ['a' => 3, 'b' => 'item 3']: the item 2, skipped by branch_a, is dropped +``` + Notes ----- -In iterable processes, inputs of different iterations may get mixed (for instance when a parent task skips an item -because of an error), even with `clean_input_on_override: true`. Its usage is **strongly discouraged** outside of -non-iterable processes; prefer a blocking task such as [ArrayMergeTask](array_merge_task.md) or -[AggregateIterableTask](aggregate_iterable_task.md) when possible. +The inputs of an item are told apart by the executions of the ancestors they have in common (their execution count, +`ProcessState::getExecutionCount()`): when an input arrives after a common ancestor has been executed again, the inputs +received from the other branches for a previous item are dropped (except those of `keep_inputs`). An item skipped by a +branch (filter, error with `error_strategy: skip`) is therefore never joined with the next item. Before v5.4, the +inputs of two items could be joined in that case. + +Inputs without any common ancestor (e.g. two branches starting with their own root task) are always joined. diff --git a/src/Manager/ProcessManager.php b/src/Manager/ProcessManager.php index 9b02d16a..ba9ce711 100644 --- a/src/Manager/ProcessManager.php +++ b/src/Manager/ProcessManager.php @@ -413,6 +413,7 @@ protected function processExecution(TaskConfiguration $taskConfiguration, int $e return; } + $state->incrementExecutionCount(); try { if (self::EXECUTE_PROCESS === $executionFlag) { $state->reset(false); diff --git a/src/Model/ProcessState.php b/src/Model/ProcessState.php index bfd691b1..af1f2605 100644 --- a/src/Model/ProcessState.php +++ b/src/Model/ProcessState.php @@ -78,6 +78,11 @@ class ProcessState protected string $status = self::STATUS_NEW; + /** + * Number of executions of the task (execute, proceed or flush), not reset between the items: tells the items apart. + */ + protected int $executionCount = 0; + public function __construct( protected ProcessConfiguration $processConfiguration, protected ProcessHistory $processHistory, @@ -258,6 +263,23 @@ public function setPreviousState(?self $previousState): void $this->previousState = $previousState; } + /** + * Number of executions of the task (execute, proceed or flush) since the beginning of the process: each output of a + * task follows an execution, so a task downstream can tell whether two inputs come from the same item. + */ + public function getExecutionCount(): int + { + return $this->executionCount; + } + + /** + * @internal called by the ProcessManager before each execution of the task + */ + public function incrementExecutionCount(): void + { + ++$this->executionCount; + } + public function getStatus(): string { return $this->status; diff --git a/src/Task/InputAggregatorTask.php b/src/Task/InputAggregatorTask.php index 7fffe702..4bb151d1 100644 --- a/src/Task/InputAggregatorTask.php +++ b/src/Task/InputAggregatorTask.php @@ -18,24 +18,29 @@ use Symfony\Component\OptionsResolver\OptionsResolver; /** - * Wait for defined inputs before passing an aggregated output. - * Should have been a BlockingTask, but due to limitations in the current model, it's a hack using skips and finalize. + * Joins the outputs of several branches: waits for an input from every parent task declared in input_codes, then + * outputs them indexed by their destination key. Unlike a blocking task, it outputs once per item. * - * @see README.md:Known issues - * @deprecated since 5.4, it's way too much error prone - should be refactored as a blocking task, will be removed in v6.0 + * The inputs of an item are told apart by the executions of their common ancestors: when a branch skips an item, the + * inputs already received for this item are dropped when the inputs of the next item arrive, instead of being joined + * with them. + * + * @final since 5.4 */ class InputAggregatorTask extends AbstractConfigurableTask { - public function __construct() - { - trigger_deprecation('cleverage/process-bundle', '5.4', 'The "%s" task is deprecated: it is too error-prone. It will be removed in v6.0.', self::class); - } - /** * @var array */ protected array $inputs = []; + /** + * Execution count of the ancestors of each received input (ancestor state id => execution count), by input code. + * + * @var array> + */ + protected array $inputAncestors = []; + /** * Store inputs and once everything has been received, pass to next task * Once an output has been generated this task is reset, and may wait for another loop. @@ -48,15 +53,19 @@ public function execute(ProcessState $state): void } $inputCode = $this->getInputCode($state); + $ancestors = $this->getAncestorExecutions($previousState); + $this->removeInputsOfPreviousItems($state, $inputCode, $ancestors); if (\array_key_exists($inputCode, $this->inputs)) { if ($this->getOption($state, 'clean_input_on_override')) { $this->inputs = []; + $this->inputAncestors = []; } else { throw new \UnexpectedValueException("The output from input '{$inputCode}' has already been defined, please use an aggregator if you have an iterable output"); } } $this->inputs[$inputCode] = $state->getInput(); + $this->inputAncestors[$inputCode] = $ancestors; if ($this->isResolved($state)) { $state->setOutput($this->inputs); @@ -66,7 +75,7 @@ public function execute(ProcessState $state): void if (null !== $keepInputs && \in_array($inputCode, $keepInputs, true)) { continue; } - unset($this->inputs[$inputCode]); + unset($this->inputs[$inputCode], $this->inputAncestors[$inputCode]); } } else { $state->setSkipped(true); @@ -118,4 +127,45 @@ protected function isResolved(ProcessState $state): bool return true; } + + /** + * Execution count of the input's ancestors, from its parent task up to the first task of its branch. + * + * @return array ancestor state id => execution count + */ + protected function getAncestorExecutions(ProcessState $previousState): array + { + $ancestors = []; + for ($ancestor = $previousState; $ancestor instanceof ProcessState; $ancestor = $ancestor->getPreviousState()) { + $id = spl_object_id($ancestor); + if (\array_key_exists($id, $ancestors)) { + break; + } + $ancestors[$id] = $ancestor->getExecutionCount(); + } + + return $ancestors; + } + + /** + * Removes the inputs of the other branches received for a previous item: a common ancestor has been executed again + * since they were received (e.g. a branch skipped this item). The inputs of keep_inputs are kept. + * + * @param array $ancestors + */ + protected function removeInputsOfPreviousItems(ProcessState $state, string $inputCode, array $ancestors): void + { + $keepInputs = $this->getOption($state, 'keep_inputs') ?? []; + foreach ($this->inputAncestors as $receivedInputCode => $receivedAncestors) { + if ($receivedInputCode === $inputCode || \in_array($receivedInputCode, $keepInputs, true)) { + continue; + } + foreach (array_intersect_key($receivedAncestors, $ancestors) as $id => $executionCount) { + if ($executionCount !== $ancestors[$id]) { + unset($this->inputs[$receivedInputCode], $this->inputAncestors[$receivedInputCode]); + break; + } + } + } + } } diff --git a/tests/App/TestKernel.php b/tests/App/TestKernel.php index bdda2eee..96d29577 100644 --- a/tests/App/TestKernel.php +++ b/tests/App/TestKernel.php @@ -111,6 +111,51 @@ protected function configureContainer(ContainerConfigurator $container): void ], ], ], + // branch_a skips the item 2, branch_b the item 4 + 'test.input_aggregator' => [ + 'end_point' => 'collect', + 'tasks' => [ + 'entry' => [ + 'task' => 'constant_iterable_output', + 'options' => ['output' => [1, 2, 3, 4, 5]], + 'outputs' => ['branch_a', 'branch_b'], + ], + 'branch_a' => ['task' => 'filter', 'options' => ['not_match' => ['' => 2]], 'outputs' => ['format_a']], + 'format_a' => ['task' => 'transformer', 'options' => ['transformers' => ['sprintf' => ['format' => 'a%s']]], 'outputs' => ['aggregate']], + 'branch_b' => ['task' => 'filter', 'options' => ['not_match' => ['' => 4]], 'outputs' => ['format_b']], + 'format_b' => ['task' => 'transformer', 'options' => ['transformers' => ['sprintf' => ['format' => 'b%s']]], 'outputs' => ['aggregate']], + 'aggregate' => [ + 'task' => 'input_aggregator', + 'options' => ['input_codes' => ['format_a' => 'a', 'format_b' => 'b']], + 'outputs' => ['collect'], + ], + 'collect' => ['task' => 'aggregate_iterable'], + ], + ], + // branch_a fails on the item "x" (intdiv() of a string), and skips it + 'test.input_aggregator_error' => [ + 'end_point' => 'collect', + 'tasks' => [ + 'entry' => [ + 'task' => 'constant_iterable_output', + 'options' => ['output' => [1, 'x', 3]], + 'outputs' => ['branch_a', 'branch_b'], + ], + 'branch_a' => [ + 'task' => 'transformer', + 'options' => ['transformers' => ['callback' => ['callback' => 'intdiv', 'right_parameters' => [1]]]], + 'error_strategy' => 'skip', + 'outputs' => ['aggregate'], + ], + 'branch_b' => ['task' => 'transformer', 'options' => ['transformers' => ['sprintf' => ['format' => 'b%s']]], 'outputs' => ['aggregate']], + 'aggregate' => [ + 'task' => 'input_aggregator', + 'options' => ['input_codes' => ['branch_a' => 'a', 'branch_b' => 'b']], + 'outputs' => ['collect'], + ], + 'collect' => ['task' => 'aggregate_iterable'], + ], + ], ], ]); diff --git a/tests/Functional/KernelTest.php b/tests/Functional/KernelTest.php index 2688dea2..fb7968af 100644 --- a/tests/Functional/KernelTest.php +++ b/tests/Functional/KernelTest.php @@ -19,9 +19,9 @@ use CleverAge\ProcessBundle\Registry\TaskRegistry; use CleverAge\ProcessBundle\Registry\TransformerRegistry; use CleverAge\ProcessBundle\Tests\App\TestKernel; -use CleverAge\ProcessBundle\Tests\IgnoreDeprecationsTrait; use CleverAge\ProcessBundle\Transformer\TransformerInterface; use PHPUnit\Framework\Attributes\CoversNothing; +use PHPUnit\Framework\Attributes\DataProvider; use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase; /** @@ -31,8 +31,6 @@ #[CoversNothing] class KernelTest extends KernelTestCase { - use IgnoreDeprecationsTrait; - private mixed $exceptionHandler; protected function setUp(): void @@ -60,12 +58,7 @@ public function testEveryTaskIsResolvedByItsCode(): void $codes = TestKernel::getTaskCodes(); self::assertNotEmpty($codes); foreach ($codes as $code) { - $taskConfiguration = $processConfiguration->getTaskConfiguration($code); - // Deprecated InputAggregatorTask, see InputAggregatorTaskTest::testTaskIsDeprecated() - $task = 'input_aggregator' === $code - ? self::ignoreDeprecations(static fn (): TaskInterface => $taskRegistry->getTask($taskConfiguration)) - : $taskRegistry->getTask($taskConfiguration); - self::assertInstanceOf(TaskInterface::class, $task, $code); + self::assertInstanceOf(TaskInterface::class, $taskRegistry->getTask($processConfiguration->getTaskConfiguration($code)), $code); } } @@ -88,6 +81,27 @@ public function testProcessIsExecuted(): void self::assertSame('Hello world', $processManager->execute('test.run')); } + /** + * @return iterable>}> + */ + public static function provideInputAggregatorProcesses(): iterable + { + yield 'filtered items' => ['test.input_aggregator', [['a' => 'a1', 'b' => 'b1'], ['a' => 'a3', 'b' => 'b3'], ['a' => 'a5', 'b' => 'b5']]]; + yield 'error with the skip strategy' => ['test.input_aggregator_error', [['a' => 1, 'b' => 'b1'], ['a' => 3, 'b' => 'b3']]]; + } + + /** + * @param list> $expected + */ + #[DataProvider('provideInputAggregatorProcesses')] + public function testInputAggregatorDropsTheItemsSkippedByABranch(string $processCode, array $expected): void + { + $processManager = self::getContainer()->get('test.process_manager'); + self::assertInstanceOf(ProcessManager::class, $processManager); + + self::assertSame($expected, $processManager->execute($processCode)); + } + private function getExceptionHandler(): mixed { $handler = set_exception_handler(null); diff --git a/tests/Task/InputAggregatorTaskTest.php b/tests/Task/InputAggregatorTaskTest.php index f4cb2014..1bda216b 100644 --- a/tests/Task/InputAggregatorTaskTest.php +++ b/tests/Task/InputAggregatorTaskTest.php @@ -21,7 +21,6 @@ use CleverAge\ProcessBundle\Model\ProcessState; use CleverAge\ProcessBundle\Task\DummyTask; use CleverAge\ProcessBundle\Task\InputAggregatorTask; -use CleverAge\ProcessBundle\Tests\IgnoreDeprecationsTrait; use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\TestCase; use Symfony\Component\OptionsResolver\Exception\InvalidOptionsException; @@ -36,33 +35,11 @@ #[\PHPUnit\Framework\Attributes\UsesClass(ProcessState::class)] class InputAggregatorTaskTest extends TestCase { - use IgnoreDeprecationsTrait; - private const INPUT_CODES = ['branch_a' => 'a', 'branch_b' => 'b']; - public function testTaskIsDeprecated(): void - { - $deprecations = []; - set_error_handler(static function (int $errno, string $errstr) use (&$deprecations): bool { - $deprecations[] = $errstr; - - return true; - }, \E_USER_DEPRECATED); - try { - new InputAggregatorTask(); - } finally { - restore_error_handler(); - } - - self::assertSame( - ['Since cleverage/process-bundle 5.4: The "'.InputAggregatorTask::class.'" task is deprecated: it is too error-prone. It will be removed in v6.0.'], - $deprecations - ); - } - public function testTaskIsSkippedUntilEveryInputIsReceived(): void { - $task = $this->createTask(); + $task = new InputAggregatorTask(); $options = ['input_codes' => self::INPUT_CODES]; $state = $this->receive($task, $options, 'branch_a', 'A'); @@ -76,7 +53,7 @@ public function testTaskIsSkippedUntilEveryInputIsReceived(): void public function testBufferIsClearedAfterOutput(): void { - $task = $this->createTask(); + $task = new InputAggregatorTask(); $options = ['input_codes' => self::INPUT_CODES]; $this->receive($task, $options, 'branch_a', 'A1'); @@ -91,7 +68,7 @@ public function testBufferIsClearedAfterOutput(): void public function testOverriddenInputClearsBufferByDefault(): void { - $task = $this->createTask(); + $task = new InputAggregatorTask(); $options = ['input_codes' => ['branch_a' => 'a', 'branch_b' => 'b', 'branch_c' => 'c']]; $this->receive($task, $options, 'branch_a', 'A1'); @@ -108,7 +85,7 @@ public function testOverriddenInputClearsBufferByDefault(): void public function testOverriddenInputThrowsWhenCleaningIsDisabled(): void { - $task = $this->createTask(); + $task = new InputAggregatorTask(); $options = ['input_codes' => self::INPUT_CODES, 'clean_input_on_override' => false]; $this->receive($task, $options, 'branch_a', 'A1'); @@ -121,7 +98,7 @@ public function testOverriddenInputThrowsWhenCleaningIsDisabled(): void public function testKeptInputsAreReusedForNextOutputs(): void { - $task = $this->createTask(); + $task = new InputAggregatorTask(); $options = ['input_codes' => self::INPUT_CODES, 'keep_inputs' => ['a']]; $this->receive($task, $options, 'branch_a', 'A'); @@ -135,7 +112,7 @@ public function testKeptInputsAreReusedForNextOutputs(): void public function testSeveralParentsCanTargetTheSameKey(): void { - $task = $this->createTask(); + $task = new InputAggregatorTask(); $options = ['input_codes' => ['branch_a' => 'value', 'branch_b' => 'value']]; $state = $this->receive($task, $options, 'branch_b', 'B'); @@ -145,7 +122,7 @@ public function testSeveralParentsCanTargetTheSameKey(): void public function testTaskWithoutPreviousStateThrows(): void { - $task = $this->createTask(); + $task = new InputAggregatorTask(); $state = $this->createState(['input_codes' => self::INPUT_CODES]); $task->initialize($state); @@ -157,7 +134,7 @@ public function testTaskWithoutPreviousStateThrows(): void public function testUnmappedPreviousTaskThrows(): void { - $task = $this->createTask(); + $task = new InputAggregatorTask(); $this->expectException(\UnexpectedValueException::class); $this->expectExceptionMessage("Task 'unknown' is not mapped in the input_codes option"); @@ -167,13 +144,61 @@ public function testUnmappedPreviousTaskThrows(): void public function testInputCodesOptionIsRequired(): void { - $task = $this->createTask(); + $task = new InputAggregatorTask(); $this->expectException(MissingOptionsException::class); $task->initialize($this->createState([])); } + public function testInputsOfAnItemSkippedByTheFirstBranchAreDropped(): void + { + $task = new InputAggregatorTask(); + $options = ['input_codes' => self::INPUT_CODES]; + [$entry, $branchA, $branchB] = $this->createBranches(); + + self::assertSame(['a' => 'a1', 'b' => 'b1'], $this->receiveItem($task, $options, $entry, [[$branchA, 'a1'], [$branchB, 'b1']])); + // The item 2 is skipped by branch_a: b2 waits for a2 + self::assertNull($this->receiveItem($task, $options, $entry, [[$branchB, 'b2']])); + // b2 is not joined with a3 + self::assertSame(['a' => 'a3', 'b' => 'b3'], $this->receiveItem($task, $options, $entry, [[$branchA, 'a3'], [$branchB, 'b3']])); + } + + public function testInputsOfAnItemSkippedByTheLastBranchAreDropped(): void + { + $task = new InputAggregatorTask(); + $options = ['input_codes' => self::INPUT_CODES]; + [$entry, $branchA, $branchB] = $this->createBranches(); + + self::assertNull($this->receiveItem($task, $options, $entry, [[$branchA, 'a1']])); + self::assertSame(['a' => 'a2', 'b' => 'b2'], $this->receiveItem($task, $options, $entry, [[$branchA, 'a2'], [$branchB, 'b2']])); + } + + public function testKeptInputsAreJoinedWithTheNextItems(): void + { + $task = new InputAggregatorTask(); + $options = ['input_codes' => self::INPUT_CODES, 'keep_inputs' => ['a']]; + [$entry, $branchA, $branchB] = $this->createBranches(); + + self::assertSame(['a' => 'a1', 'b' => 'b1'], $this->receiveItem($task, $options, $entry, [[$branchA, 'a1'], [$branchB, 'b1']])); + self::assertSame(['a' => 'a1', 'b' => 'b2'], $this->receiveItem($task, $options, $entry, [[$branchB, 'b2']])); + } + + public function testInputsWithoutCommonAncestorAreJoined(): void + { + $task = new InputAggregatorTask(); + $options = ['input_codes' => self::INPUT_CODES]; + [, $branchA] = $this->createBranches(); + [, , $branchB] = $this->createBranches(); + + $branchA->incrementExecutionCount(); + $this->receiveFrom($task, $options, $branchA, 'a1'); + $branchA->incrementExecutionCount(); + $branchB->incrementExecutionCount(); + + self::assertSame(['a' => 'a1', 'b' => 'b1'], $this->receiveFrom($task, $options, $branchB, 'b1')->getOutput()); + } + /** * @return iterable}> */ @@ -190,7 +215,7 @@ public static function invalidOptionsProvider(): iterable #[DataProvider('invalidOptionsProvider')] public function testInvalidOptionsAreRejected(array $options): void { - $task = $this->createTask(); + $task = new InputAggregatorTask(); $this->expectException(InvalidOptionsException::class); @@ -213,9 +238,61 @@ private function receive(InputAggregatorTask $task, array $options, string $prev return $state; } - private function createTask(): InputAggregatorTask + /** + * An entry task whose outputs go to branch_a and branch_b. + * + * @return array{ProcessState, ProcessState, ProcessState} + */ + private function createBranches(): array + { + $entry = $this->createState([], 'entry'); + $branchA = $this->createState([], 'branch_a'); + $branchA->setPreviousState($entry); + $branchB = $this->createState([], 'branch_b'); + $branchB->setPreviousState($entry); + + return [$entry, $branchA, $branchB]; + } + + /** + * A new item of the entry task, sent to the branches that do not skip it (in the order of the outputs). + * + * @param array $options + * @param list $inputs branch state, input + * + * @return array|null the output of the task after the last input, null if it is skipped + */ + private function receiveItem(InputAggregatorTask $task, array $options, ProcessState $entry, array $inputs): ?array + { + $entry->incrementExecutionCount(); + $state = null; + foreach ($inputs as [$branch, $input]) { + $branch->incrementExecutionCount(); + $state = $this->receiveFrom($task, $options, $branch, $input); + } + self::assertInstanceOf(ProcessState::class, $state); + if ($state->isSkipped()) { + return null; + } + $output = $state->getOutput(); + self::assertIsArray($output); + + return $output; + } + + /** + * @param array $options + */ + private function receiveFrom(InputAggregatorTask $task, array $options, ProcessState $previousState, string $input): ProcessState { - return self::ignoreDeprecations(static fn (): InputAggregatorTask => new InputAggregatorTask()); + $state = $this->createState($options); + $state->setPreviousState($previousState); + $state->setInput($input); + + $task->initialize($state); + $task->execute($state); + + return $state; } /**