Add auto update count processed item while running job (#79 )

* Add auto update count processed item while running job
Added custom index for job status (#77 )
2026-03-24 14:52:21 +01:00 · 2025-10-02 15:22:37 +02:00 · 2025-07-04 09:26:51 +02:00 · 2024-10-31 16:40:09 +01:00 · 2024-10-31 15:38:10 +01:00 · 2024-10-31 13:01:43 +01:00
65 changed files with 753 additions and 587 deletions
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -0,0 +1,27 @@
+name: Build
+
+on:
+  push:
+    branches:
+      - master
+
+
+jobs:
+  build:
+    name: Build
+    runs-on: ubuntu-latest
+    permissions: read-all
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          fetch-depth: 0  # Shallow clones should be disabled for a better relevancy of analysis
+      - uses: sonarsource/sonarqube-scan-action@master
+        env:
+          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
+          SONAR_HOST_URL: ${{ secrets.SONAR_HOST_URL }}
+      # If you wish to fail your job when the Quality Gate is red, uncomment the
+      # following lines. This would typically be used to fail a deployment.
+      # - uses: sonarsource/sonarqube-quality-gate-action@master
+      #   timeout-minutes: 5
+      #   env:
+      #     SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -0,0 +1,14 @@
+name: CI
+
+on: [push]
+
+jobs:
+  build-test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+      - uses: php-actions/composer@v6 # or alternative dependency management
+      - uses: php-actions/phpunit@v4
+      - name: Run PHP CS Fixer
+        run: php vendor/bin/php-cs-fixer fix --dry-run --diff
--- a/.github/workflows/semicolons-kudos.yaml
+++ b/.github/workflows/semicolons-kudos.yaml
@@ -0,0 +1,26 @@
+
+name: Kudos for Code
+on:
+  push:
+    branches: ["master"]
+  workflow_dispatch:
+
+jobs:
+  kudos:
+    name: Semicolons Kudos
+    permissions: write-all
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+      - uses: LoremLabs/kudos-for-code-action@latest
+        with:
+          search-dir: "."
+          destination: "artifact"
+          generate-nomerges: true
+          generate-validemails: true
+          generate-limitdepth: 0
+          generate-fromrepo: true
+          analyze-repo: false
+          skip-ids: ""
--- a/.gitignore
+++ b/.gitignore
@@ -3,3 +3,7 @@ composer.lock
 .phpunit.result.cache
 .php_cs.cache
 .php_cs
+.idea
+.phpunit.cache
+.php-version
+.php-cs-fixer.cache
--- a/.php-cs-fixer.dist.php
+++ b/.php-cs-fixer.dist.php
@@ -0,0 +1,49 @@
+<?php
+
+$finder = PhpCsFixer\Finder::create()
+    ->in('src')
+    //->in('tests')
+    ->files()->name('*.php');
+
+$config = new PhpCsFixer\Config();
+$config->setRules([
+    '@Symfony' => true,
+    '@Symfony:risky' => true,
+    '@PSR12' => true,
+    'array_syntax' => [
+        'syntax' => 'short',
+    ],
+    'combine_consecutive_unsets' => true,
+    'native_function_invocation' => [
+        'include' => [
+            '@compiler_optimized',
+        ],
+    ],
+    'no_extra_blank_lines' => [
+        'tokens' => [
+            'break',
+            'continue',
+            'extra',
+            'return',
+            'throw',
+            'use',
+            'parenthesis_brace_block',
+            'square_brace_block',
+            'curly_brace_block',
+        ],
+    ],
+    'ordered_class_elements' => true,
+    'ordered_imports' => true,
+    'yoda_style' => [
+        'equal' => false,
+        'identical' => false,
+        'less_and_greater' => false,
+        'always_move_variable' => false,
+    ],
+])
+    ->setRiskyAllowed(true)
+    ->setFinder(
+        $finder
+    );
+
+return $config;
--- a/.php_cs.dist
+++ b/.php_cs.dist
@@ -1,12 +0,0 @@
-<?php
-
-$finder = PhpCsFixer\Finder::create()->in(__DIR__.'/src');
-
-return PhpCsFixer\Config::create()
-    ->setRules([
-        '@Symfony' => true,
-        'declare_strict_types' => true,
-    ])
-    ->setFinder($finder)
-    ->setRiskyAllowed(true)
-;
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,31 @@
+# Version 5.3.0
+* Added auto update count processed item while running job
+
+# Version 5.2.0
+* Added custom index for job status
+
+# Version 5.1.0
+* Refactor SchemaDump command
+
+# Version 5.0.1
+* Fix compatibility with doctrine 4
+
+# Version 5.0.0
+* Initiate Kudos on dataflow-bundle
+* Added Symfony 7 support
+* Removed Symfony 6 compatibility
+* Removed Symfony 5 compatibility
+* Removed Symfony 4 compatibility
+* Removed Symfony 3 compatibility
+* Changed README.md
+* Added CI
+
+# Version 4.1.3
+* Fix log exception argument typing
+
+# Version 4.1.2
+* Fix DBAL 2.12 compatibility break
+
 # Version 4.1.0

 * Added custom index for exception log
--- a/README.md
+++ b/README.md
@@ -1,14 +1,23 @@
 # Code Rhapsodie Dataflow Bundle

-DataflowBundle is a bundle for Symfony 3.4+ 
+DataflowBundle is a bundle for Symfony 3.4+
 providing an easy way to create import / export dataflow.

-Dataflow uses a linear generic workflow in three parts:
- * one reader
- * any number of steps that can be synchronous or asynchronous
- * one or more writers
+| Dataflow | Symfony                  | Support |
+|----------|--------------------------|---------|
+| 5.x      | 7.x                      | yes     |
+| 4.x      | 3.4 \| 4.x \| 5.x \| 6.x | yes     |
+| 3.x      | 3.4 \| 4.x \| 5.x        | no      |
+| 2.x      | 3.4 \| 4.x               | no      |
+| 1.x      | 3.4 \| 4.x               | no      |

-The reader can read data from anywhere and return data row by row. Each step processes the current row data. 
+Dataflow uses a linear generic workflow in three parts:
+
+* one reader
+* any number of steps that can be synchronous or asynchronous
+* one or more writers
+
+The reader can read data from anywhere and return data row by row. Each step processes the current row data.
 The steps are executed in the order in which they are added.
 And, one or more writers save the row anywhere you want.

@@ -27,7 +36,6 @@ As the following schema shows, you can define more than one dataflow:
 * Display the result for the last Job for a Dataflow from the command line
 * Work with multiple Doctrine DBAL connections

-
 ## Installation

 Security notice: Symfony 4.x is not supported before 4.1.12, see https://github.com/advisories/GHSA-pgwj-prpq-jpc2
@@ -44,7 +52,8 @@ $ composer require code-rhapsodie/dataflow-bundle

 You can use the generic readers, writers and steps from [PortPHP](https://github.com/portphp/portphp).

-For the writers, you must use the adapter `CodeRhapsodie\DataflowBundle\DataflowType\Writer\PortWriterAdapter` like this:
+For the writers, you must use the adapter `CodeRhapsodie\DataflowBundle\DataflowType\Writer\PortWriterAdapter` like
+this:

 ```php
 <?php
@@ -57,9 +66,7 @@ $builder->addWriter(new \CodeRhapsodie\DataflowBundle\DataflowType\Writer\PortWr

 ### Register the bundle

-#### Symfony 4 (new tree)
-
-For Symfony 4, add `CodeRhapsodie\DataflowBundle\CodeRhapsodieDataflowBundle::class => ['all' => true],
+Add `CodeRhapsodie\DataflowBundle\CodeRhapsodieDataflowBundle::class => ['all' => true],
 ` in the `config/bundles.php` file.

 Like this:
@@ -74,32 +81,13 @@ return [
 ];
 ```

-#### Symfony 3.4 (old tree)
-
-For Symfony 3.4, add a new line in the `app/AppKernel.php` file.
-
-Like this:
-
-```php
-<?php
-// app/AppKernel.php
-
-public function registerBundles()
-{
-    $bundles = [
-        // ...
-        new CodeRhapsodie\DataflowBundle\CodeRhapsodieDataflowBundle(),
-        // ...
-    ];
-}
-```
-
 ### Update the database

 This bundle uses Doctrine DBAL to store Dataflow schedule into the database table (`cr_dataflow_scheduled`)
 and jobs (`cr_dataflow_job`).

-If you use [Doctrine Migration Bundle](https://symfony.com/doc/master/bundles/DoctrineMigrationsBundle/index.html) or [Phinx](https://phinx.org/) 
+If you use [Doctrine Migration Bundle](https://symfony.com/doc/master/bundles/DoctrineMigrationsBundle/index.html)
+or [Phinx](https://phinx.org/)
 or [Kaliop Migration Bundle](https://github.com/kaliop-uk/ezmigrationbundle) or whatever,
 you can add a new migration with the generated SQL query from this command:

@@ -137,6 +125,7 @@ Dataflow can delegate the execution of its jobs to the Symfony messenger compone
 This allows jobs to be executed concurrently by workers instead of sequentially.

 To enable messenger mode:
+
 ```yaml
 code_rhapsodie_dataflow:
  messenger_mode:
@@ -145,6 +134,7 @@ code_rhapsodie_dataflow:
 ```

 You also need to route Dataflow messages to the proper transport:
+
 ```yaml
 # config/packages/messenger.yaml
 framework:
@@ -158,9 +148,11 @@ framework:

 ## Define a dataflow type

-This bundle uses a fixed and simple workflow structure in order to let you focus on the data processing logic part of your dataflow.
+This bundle uses a fixed and simple workflow structure in order to let you focus on the data processing logic part of
+your dataflow.

 A dataflow type defines the different parts of your dataflow. A dataflow is made of:
+
 - exactly one *Reader*
 - any number of *Steps*
 - one or more *Writers*
@@ -169,8 +161,10 @@ Dataflow types can be configured with options.

 A dataflow type must implement `CodeRhapsodie\DataflowBundle\DataflowType\DataflowTypeInterface`.

-To help with creating your dataflow types, an abstract class `CodeRhapsodie\DataflowBundle\DataflowType\AbstractDataflowType`
-is provided, allowing you to define your dataflow through a handy builder `CodeRhapsodie\DataflowBundle\DataflowType\DataflowBuilder`.
+To help with creating your dataflow types, an abstract
+class `CodeRhapsodie\DataflowBundle\DataflowType\AbstractDataflowType`
+is provided, allowing you to define your dataflow through a handy
+builder `CodeRhapsodie\DataflowBundle\DataflowType\DataflowBuilder`.

 This is an example to define one class DataflowType:

@@ -230,15 +224,16 @@ class MyFirstDataflowType extends AbstractDataflowType

 Dataflow types must be tagged with `coderhapsodie.dataflow.type`.

-If you're using Symfony auto-configuration for your services, this tag will be automatically added to all services implementing `DataflowTypeInterface`.
+If you're using Symfony auto-configuration for your services, this tag will be automatically added to all services
+implementing `DataflowTypeInterface`.

 Otherwise, manually add the tag `coderhapsodie.dataflow.type` in your dataflow type service configuration:

 ```yaml
 ```yaml
-    CodeRhapsodie\DataflowExemple\DataflowType\MyFirstDataflowType:
-      tags:
-        - { name: coderhapsodie.dataflow.type }
+CodeRhapsodie\DataflowExemple\DataflowType\MyFirstDataflowType:
+  tags:
+    - { name: coderhapsodie.dataflow.type }
 ```

 ### Use options for your dataflow type
@@ -264,11 +259,14 @@ class MyFirstDataflowType extends AbstractDataflowType
 }
 ```

-With this configuration, the option `fileName` is required. For an advanced usage of the option resolver, read the [Symfony documentation](https://symfony.com/doc/current/components/options_resolver.html).
+With this configuration, the option `fileName` is required. For an advanced usage of the option resolver, read
+the [Symfony documentation](https://symfony.com/doc/current/components/options_resolver.html).

 For asynchronous management, `AbstractDataflowType` come with two default options :
+
 - loopInterval : default to 0. Update this interval if you wish customise the `tick` loop duration.
- emitInterval : default to 0. Update this interval to have a control when reader must emit new data in the flow pipeline.
+- emitInterval : default to 0. Update this interval to have a control when reader must emit new data in the flow
+  pipeline.

 ### Logging

@@ -292,14 +290,15 @@ class MyDataflowType extends AbstractDataflowType
 }
 ```

-When using the `code-rhapsodie:dataflow:run-pending` command, this logger will also be used to save the log in the corresponding job in the database.
+When using the `code-rhapsodie:dataflow:run-pending` command, this logger will also be used to save the log in the
+corresponding job in the database.

 ### Check if your DataflowType is ready

 Execute this command to check if your DataflowType is correctly registered:

 ```shell script
-$ bin/console debug:container --tag coderhapsodie.dataflow.type --show-private
+$ bin/console debug:container --tag coderhapsodie.dataflow.type
 ```

 The result is like this:
@@ -316,10 +315,10 @@ Symfony Container Public and Private Services Tagged with "coderhapsodie.dataflo

 ```

-
 ### Readers

-*Readers* provide the dataflow with elements to import / export. Usually, elements are read from an external resource (file, database, webservice, etc).
+*Readers* provide the dataflow with elements to import / export. Usually, elements are read from an external resource (
+file, database, webservice, etc).

 A *Reader* can be any `iterable`.

@@ -357,15 +356,16 @@ You can set up this reader as follows:
 $builder->setReader(($this->myReader)())
 ``` 

-
 ### Steps

 *Steps* are operations performed on the elements before they are handled by the *Writers*. Usually, steps are either:
+
 - converters, that alter the element
 - filters, that conditionally prevent further operations on the element
 - generators, that can include asynchronous operations

 A *Step* can be any callable, taking the element as its argument, and returning either:
+
 - the element, possibly altered
 - `false`, if no further operations should be performed on this element

@@ -409,7 +409,8 @@ Note : you can ensure writing order for asynchronous operations if all steps are
 *Writers* perform the actual import / export operations.

 A *Writer* must implement `CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface`.
-As this interface is not compatible with `Port\Writer`, the adapter `CodeRhapsodie\DataflowBundle\DataflowType\Writer\PortWriterAdapter` is provided.
+As this interface is not compatible with `Port\Writer`, the
+adapter `CodeRhapsodie\DataflowBundle\DataflowType\Writer\PortWriterAdapter` is provided.

 This example show how to use the predefined PhpPort Writer :

@@ -460,7 +461,9 @@ class FileWriter implements WriterInterface

 #### CollectionWriter

-If you want to write multiple items from a single item read, you can use the generic `CollectionWriter`. This writer will iterate over any `iterable` it receives, and pass each item from that collection to your own writer that handles single items.
+If you want to write multiple items from a single item read, you can use the generic `CollectionWriter`. This writer
+will iterate over any `iterable` it receives, and pass each item from that collection to your own writer that handles
+single items.

 ```php
 $builder->addWriter(new CollectionWriter($mySingleItemWriter));
@@ -470,9 +473,11 @@ $builder->addWriter(new CollectionWriter($mySingleItemWriter));

 If you want to call different writers depending on what item is read, you can use the generic `DelegatorWriter`.

-As an example, let's suppose our items are arrays with the first entry being either `product` or `order`. We want to use a different writer based on that value.
+As an example, let's suppose our items are arrays with the first entry being either `product` or `order`. We want to use
+a different writer based on that value.

-First, create your writers implementing `DelegateWriterInterface` (this interface extends `WriterInterface` so your writers can still be used without the `DelegatorWriter`).
+First, create your writers implementing `DelegateWriterInterface` (this interface extends `WriterInterface` so your
+writers can still be used without the `DelegatorWriter`).

 ```php
 <?php
@@ -545,7 +550,8 @@ Then, configure your `DelegatorWriter` and add it to your dataflow type.
    }
 ```

-During execution, the `DelegatorWriter` will simply pass each item received to its first delegate (in the order those were added) that supports it. If no delegate supports an item, an exception will be thrown.
+During execution, the `DelegatorWriter` will simply pass each item received to its first delegate (in the order those
+were added) that supports it. If no delegate supports an item, an exception will be thrown.

 ## Queue

@@ -563,7 +569,8 @@ Several commands are provided to manage schedules and run jobs.

 `code-rhapsodie:dataflow:run-pending` Executes job in the queue according to their schedule.

-When messenger mode is enabled, jobs will still be created according to their schedule, but execution will be handled by the messenger component instead.
+When messenger mode is enabled, jobs will still be created according to their schedule, but execution will be handled by
+the messenger component instead.

 `code-rhapsodie:dataflow:schedule:list` Display the list of dataflows scheduled.

@@ -602,7 +609,7 @@ $ bin/console code-rhapsodie:dataflow:run-pending --connection=default

 Please report issues and request features at https://github.com/code-rhapsodie/dataflow-bundle/issues.

-Please note that only the last release of the 3.x and the 4.x versions of this bundle are actively supported.
+Please note that only the last release of the 4.x and the 5.x versions of this bundle are actively supported.

 # Contributing

--- a/Tests/DataflowType/Dataflow/AMPAsyncDataflowTest.php
+++ b/Tests/DataflowType/Dataflow/AMPAsyncDataflowTest.php
@@ -4,7 +4,6 @@ namespace CodeRhapsodie\DataflowBundle\Tests\DataflowType\Dataflow;

 use Amp\Delayed;
 use CodeRhapsodie\DataflowBundle\DataflowType\Dataflow\AMPAsyncDataflow;
-use CodeRhapsodie\DataflowBundle\DataflowType\Dataflow\Dataflow;
 use CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface;
 use PHPUnit\Framework\TestCase;

--- a/Tests/DataflowType/Writer/CollectionWriterTest.php
+++ b/Tests/DataflowType/Writer/CollectionWriterTest.php
@@ -40,10 +40,11 @@ class CollectionWriterTest extends TestCase
            ->expects($this->once())
            ->method('finish')
        ;
+        $matcher = $this->exactly(count($values));
        $embeddedWriter
-            ->expects($this->exactly(count($values)))
+            ->expects($matcher)
            ->method('write')
-            ->withConsecutive(...array_map(fn($item) => [$item], $values))
+            ->with($this->callback(fn($arg) => $arg === $values[$matcher->numberOfInvocations() - 1]))
        ;

        $writer = new CollectionWriter($embeddedWriter);
--- a/Tests/DataflowType/Writer/DelegatorWriterTest.php
+++ b/Tests/DataflowType/Writer/DelegatorWriterTest.php
@@ -10,13 +10,10 @@ use PHPUnit\Framework\TestCase;

 class DelegatorWriterTest extends TestCase
 {
-    private \CodeRhapsodie\DataflowBundle\DataflowType\Writer\DelegatorWriter $delegatorWriter;
-
-    private \CodeRhapsodie\DataflowBundle\DataflowType\Writer\DelegateWriterInterface|\PHPUnit\Framework\MockObject\MockObject $delegateInt;
-
-    private \CodeRhapsodie\DataflowBundle\DataflowType\Writer\DelegateWriterInterface|\PHPUnit\Framework\MockObject\MockObject $delegateString;
-
-    private \CodeRhapsodie\DataflowBundle\DataflowType\Writer\DelegateWriterInterface|\PHPUnit\Framework\MockObject\MockObject $delegateArray;
+    private DelegatorWriter $delegatorWriter;
+    private DelegateWriterInterface|MockObject $delegateInt;
+    private DelegateWriterInterface|MockObject $delegateString;
+    private DelegateWriterInterface|MockObject $delegateArray;

    protected function setUp(): void
    {
--- a/Tests/DataflowType/Writer/PortWriterAdapterTest.php
+++ b/Tests/DataflowType/Writer/PortWriterAdapterTest.php
@@ -11,8 +11,8 @@ class PortWriterAdapterTest extends TestCase
    {
        $value = 'not an array';

-        $writer = $this->getMockBuilder('\Port\Writer')
-            ->setMethods(['prepare', 'finish', 'writeItem'])
+        $writer = $this->getMockBuilder(\Port\Writer::class)
+            ->onlyMethods(['prepare', 'finish', 'writeItem'])
            ->getMock()
        ;
        $writer
--- a/Tests/Manager/ScheduledDataflowManagerTest.php
+++ b/Tests/Manager/ScheduledDataflowManagerTest.php
@@ -2,10 +2,8 @@

 namespace CodeRhapsodie\DataflowBundle\Tests\Manager;

-use CodeRhapsodie\DataflowBundle\DataflowType\DataflowTypeInterface;
 use CodeRhapsodie\DataflowBundle\Entity\Job;
 use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
-use CodeRhapsodie\DataflowBundle\Exceptions\UnknownDataflowTypeException;
 use CodeRhapsodie\DataflowBundle\Manager\ScheduledDataflowManager;
 use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
@@ -15,13 +13,10 @@ use PHPUnit\Framework\TestCase;

 class ScheduledDataflowManagerTest extends TestCase
 {
-    private \CodeRhapsodie\DataflowBundle\Manager\ScheduledDataflowManager $manager;
-
-    private \Doctrine\DBAL\Connection|\PHPUnit\Framework\MockObject\MockObject $connection;
-
-    private \CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository|\PHPUnit\Framework\MockObject\MockObject $scheduledDataflowRepository;
-
-    private \CodeRhapsodie\DataflowBundle\Repository\JobRepository|\PHPUnit\Framework\MockObject\MockObject $jobRepository;
+    private ScheduledDataflowManager $manager;
+    private Connection|MockObject $connection;
+    private ScheduledDataflowRepository|MockObject $scheduledDataflowRepository;
+    private JobRepository|MockObject $jobRepository;

    protected function setUp(): void
    {
@@ -50,10 +45,15 @@ class ScheduledDataflowManagerTest extends TestCase
            ->willReturn([$scheduled1, $scheduled2])
        ;

+        $matcher = $this->exactly(2);
        $this->jobRepository
-            ->expects($this->exactly(2))
+            ->expects($matcher)
            ->method('findPendingForScheduledDataflow')
-            ->withConsecutive([$scheduled1], [$scheduled2])
+            ->with($this->callback(fn($arg) => match ($matcher->numberOfInvocations()) {
+                1 => $arg === $scheduled1,
+                2 => $arg === $scheduled2,
+                default => false,
+            }))
            ->willReturnOnConsecutiveCalls(new Job(), null)
        ;

@@ -103,7 +103,7 @@ class ScheduledDataflowManagerTest extends TestCase
        $this->jobRepository
            ->expects($this->exactly(1))
            ->method('findPendingForScheduledDataflow')
-            ->withConsecutive([$scheduled1])
+            ->with($scheduled1)
            ->willThrowException(new \Exception())
        ;

--- a/Tests/MessengerMode/JobMessageHandlerTest.php
+++ b/Tests/MessengerMode/JobMessageHandlerTest.php
@@ -14,11 +14,9 @@ use PHPUnit\Framework\TestCase;

 class JobMessageHandlerTest extends TestCase
 {
-    private \CodeRhapsodie\DataflowBundle\Repository\JobRepository|\PHPUnit\Framework\MockObject\MockObject $repository;
-
-    private \CodeRhapsodie\DataflowBundle\Processor\JobProcessorInterface|\PHPUnit\Framework\MockObject\MockObject $processor;
-
-    private \CodeRhapsodie\DataflowBundle\MessengerMode\JobMessageHandler $handler;
+    private JobRepository|MockObject $repository;
+    private JobProcessorInterface|MockObject $processor;
+    private JobMessageHandler $handler;

    protected function setUp(): void
    {
@@ -28,11 +26,6 @@ class JobMessageHandlerTest extends TestCase
        $this->handler = new JobMessageHandler($this->repository, $this->processor);
    }

-    public function testGetHandledMessages()
-    {
-        $this->assertSame([JobMessage::class], JobMessageHandler::getHandledMessages());
-    }
-
    public function testInvoke()
    {
        $message = new JobMessage($id = 32);
--- a/Tests/Processor/JobProcessorTest.php
+++ b/Tests/Processor/JobProcessorTest.php
@@ -16,13 +16,10 @@ use Symfony\Component\EventDispatcher\EventDispatcherInterface;

 class JobProcessorTest extends TestCase
 {
-    private \CodeRhapsodie\DataflowBundle\Processor\JobProcessor $processor;
-
-    private \CodeRhapsodie\DataflowBundle\Repository\JobRepository|\PHPUnit\Framework\MockObject\MockObject $repository;
-
-    private \CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface|\PHPUnit\Framework\MockObject\MockObject $registry;
-
-    private \Symfony\Component\EventDispatcher\EventDispatcherInterface|\PHPUnit\Framework\MockObject\MockObject $dispatcher;
+    private JobProcessor $processor;
+    private JobRepository|MockObject $repository;
+    private DataflowTypeRegistryInterface|MockObject $registry;
+    private EventDispatcherInterface|MockObject $dispatcher;

    protected function setUp(): void
    {
@@ -42,36 +39,18 @@ class JobProcessorTest extends TestCase
            ->setOptions($options = ['option1' => 'value1'])
        ;

-        // Symfony 3.4 to 4.4 call
-        if (!class_exists(\Symfony\Contracts\EventDispatcher\Event::class)) {
-            $this->dispatcher
-                ->expects($this->exactly(2))
-                ->method('dispatch')
-                ->withConsecutive(
-                    [
-                        Events::BEFORE_PROCESSING,
-                        $this->callback(fn(ProcessingEvent $event) => $event->getJob() === $job)
-                    ],
-                    [
-                        Events::AFTER_PROCESSING,
-                        $this->callback(fn(ProcessingEvent $event) => $event->getJob() === $job)
-                    ],
-                );
-        } else { // Symfony 5.0+
-            $this->dispatcher
-                ->expects($this->exactly(2))
-                ->method('dispatch')
-                ->withConsecutive(
-                    [
-                        $this->callback(fn(ProcessingEvent $event) => $event->getJob() === $job),
-                        Events::BEFORE_PROCESSING,
-                    ],
-                    [
-                        $this->callback(fn(ProcessingEvent $event) => $event->getJob() === $job),
-                        Events::AFTER_PROCESSING,
-                    ],
-                );
-        }
+        $matcher = $this->exactly(2);
+        $this->dispatcher
+            ->expects($matcher)
+            ->method('dispatch')
+            ->with(
+                $this->callback(fn($arg) => $arg instanceof ProcessingEvent && $arg->getJob() === $job),
+                $this->callback(fn($arg) => match ($matcher->numberOfInvocations()) {
+                    1 => $arg === Events::BEFORE_PROCESSING,
+                    2 => $arg === Events::AFTER_PROCESSING,
+                    default => false,
+                })
+            );

        $dataflowType = $this->createMock(DataflowTypeInterface::class);

--- a/Tests/Registry/DataflowTypeRegistryTest.php
+++ b/Tests/Registry/DataflowTypeRegistryTest.php
@@ -10,7 +10,7 @@ use PHPUnit\Framework\TestCase;

 class DataflowTypeRegistryTest extends TestCase
 {
-    private \CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistry $registry;
+    private DataflowTypeRegistry $registry;

    protected function setUp(): void
    {
--- a/Tests/Runner/MessengerDataflowRunnerTest.php
+++ b/Tests/Runner/MessengerDataflowRunnerTest.php
@@ -13,11 +13,9 @@ use Symfony\Component\Messenger\MessageBusInterface;

 class MessengerDataflowRunnerTest extends TestCase
 {
-    private \CodeRhapsodie\DataflowBundle\Runner\MessengerDataflowRunner $runner;
-
-    private \CodeRhapsodie\DataflowBundle\Repository\JobRepository|\PHPUnit\Framework\MockObject\MockObject $repository;
-
-    private \Symfony\Component\Messenger\MessageBusInterface|\PHPUnit\Framework\MockObject\MockObject $bus;
+    private MessengerDataflowRunner $runner;
+    private JobRepository|MockObject $repository;
+    private MessageBusInterface|MockObject $bus;

    protected function setUp(): void
    {
@@ -37,20 +35,26 @@ class MessengerDataflowRunnerTest extends TestCase
            ->method('findNextPendingDataflow')
            ->willReturnOnConsecutiveCalls($job1, $job2, null)
        ;
+        $matcher = $this->exactly(2);
        $this->repository
-            ->expects($this->exactly(2))
+            ->expects($matcher)
            ->method('save')
-            ->withConsecutive([$job1], [$job2])
+            ->with($this->callback(fn($arg) => match ($matcher->numberOfInvocations()) {
+                1 => $arg === $job1,
+                2 => $arg === $job2,
+                default => false,
+            }))
        ;

+        $matcher = $this->exactly(2);
        $this->bus
-            ->expects($this->exactly(2))
+            ->expects($matcher)
            ->method('dispatch')
-            ->withConsecutive([
-                $this->callback(fn($message) => $message instanceof JobMessage && $message->getJobId() === $id1)
-            ], [
-                $this->callback(fn($message) => $message instanceof JobMessage && $message->getJobId() === $id2)
-            ])
+            ->with($this->callback(fn($arg) => match ($matcher->numberOfInvocations()) {
+                1 => $arg instanceof JobMessage && $arg->getJobId() === $id1,
+                2 => $arg instanceof JobMessage && $arg->getJobId() === $id2,
+                default => false,
+            }))
            ->willReturnOnConsecutiveCalls(
                new Envelope(new JobMessage($id1)),
                new Envelope(new JobMessage($id2))
--- a/Tests/Runner/PendingDataflowRunnerTest.php
+++ b/Tests/Runner/PendingDataflowRunnerTest.php
@@ -11,11 +11,9 @@ use PHPUnit\Framework\TestCase;

 class PendingDataflowRunnerTest extends TestCase
 {
-    private \CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunner $runner;
-
-    private \CodeRhapsodie\DataflowBundle\Repository\JobRepository|\PHPUnit\Framework\MockObject\MockObject $repository;
-
-    private \CodeRhapsodie\DataflowBundle\Processor\JobProcessorInterface|\PHPUnit\Framework\MockObject\MockObject $processor;
+    private PendingDataflowRunner $runner;
+    private JobRepository|MockObject $repository;
+    private JobProcessorInterface|MockObject $processor;

    protected function setUp(): void
    {
@@ -36,10 +34,15 @@ class PendingDataflowRunnerTest extends TestCase
            ->willReturnOnConsecutiveCalls($job1, $job2, null)
        ;

+        $matcher = $this->exactly(2);
        $this->processor
-            ->expects($this->exactly(2))
+            ->expects($matcher)
            ->method('process')
-            ->withConsecutive([$job1], [$job2])
+            ->with($this->callback(fn($arg) => match ($matcher->numberOfInvocations()) {
+                1 => $arg === $job1,
+                2 => $arg === $job2,
+                default => false,
+            }))
        ;

        $this->runner->runPendingDataflows();
--- a/Tests/Validator/Constraints/FrequencyValidatorTest.php
+++ b/Tests/Validator/Constraints/FrequencyValidatorTest.php
@@ -4,18 +4,18 @@ namespace CodeRhapsodie\DataflowBundle\Tests\Validator\Constraints;

 use CodeRhapsodie\DataflowBundle\Validator\Constraints\Frequency;
 use CodeRhapsodie\DataflowBundle\Validator\Constraints\FrequencyValidator;
+use PHPUnit\Framework\Attributes\DataProvider;
+use Symfony\Component\Validator\ConstraintValidatorInterface;
 use Symfony\Component\Validator\Test\ConstraintValidatorTestCase;

 class FrequencyValidatorTest extends ConstraintValidatorTestCase
 {
-    protected function createValidator()
+    protected function createValidator(): ConstraintValidatorInterface
    {
        return new FrequencyValidator();
    }

-    /**
-     * @dataProvider getValidValues
-     */
+    #[DataProvider('getValidValues')]
    public function testValidValues($value)
    {
        $this->validator->validate($value, new Frequency());
@@ -23,7 +23,7 @@ class FrequencyValidatorTest extends ConstraintValidatorTestCase
        $this->assertNoViolation();
    }

-    public function getValidValues()
+    public static function getValidValues()
    {
        return [
            ['3 days'],
@@ -47,9 +47,7 @@ class FrequencyValidatorTest extends ConstraintValidatorTestCase
        ;
    }

-    /**
-     * @dataProvider getNegativeValues
-     */
+    #[DataProvider('getNegativeValues')]
    public function testNegativeIntervals($value)
    {
        $constraint = new Frequency([
@@ -64,7 +62,7 @@ class FrequencyValidatorTest extends ConstraintValidatorTestCase
        ;
    }

-    public function getNegativeValues()
+    public static function getNegativeValues()
    {
        return [
            ['now'],
--- a/composer.json
+++ b/composer.json
@@ -43,26 +43,28 @@
    "require": {
        "php": "^8.0",
        "ext-json": "*",
-        "doctrine/dbal": "^2.12||^3.0",
-        "doctrine/doctrine-bundle": "^1.0||^2.0",
-        "monolog/monolog": "^1.0||^2.0",
+        "doctrine/dbal": "^3.0||^4.0",
+        "doctrine/doctrine-bundle": "^2.0",
+        "monolog/monolog": "^2.0||^3.0",
        "psr/log": "^1.1||^2.0||^3.0",
-        "symfony/config": "^3.4||^4.0||^5.0||^6.0",
-        "symfony/console": "^3.4||^4.0||^5.0||^6.0",
-        "symfony/dependency-injection": "^3.4||>=4.1.12||^5.0||^6.0",
-        "symfony/event-dispatcher": "^3.4||^4.0||^5.0||^6.0",
-        "symfony/http-kernel": "^3.4||^4.0||^5.0||^6.0",
-        "symfony/lock": "^3.4||^4.0||^5.0||^6.0",
-        "symfony/monolog-bridge": "^3.4||^4.0||^5.0||^6.0",
-        "symfony/options-resolver": "^3.4||^4.0||^5.0||^6.0",
-        "symfony/validator": "^3.4||^4.0||^5.0||^6.0",
-        "symfony/yaml": "^3.4||^4.0||^5.0||^6.0"
+        "symfony/config": "^7.0",
+        "symfony/console": "^7.0",
+        "symfony/dependency-injection": "^7.0",
+        "symfony/event-dispatcher": "^7.0",
+        "symfony/http-kernel": "^7.0",
+        "symfony/lock": "^7.0",
+        "symfony/monolog-bridge": "^7.0",
+        "symfony/options-resolver": "^7.0",
+        "symfony/validator": "^7.0",
+        "symfony/yaml": "^7.0"
    },
    "require-dev": {
        "amphp/amp": "^2.5",
-        "phpunit/phpunit": "^7||^8||^9",
-        "rector/rector": "^0.13.10",
-        "symfony/messenger": "^4.4||^5.0||^6.0"
+        "friendsofphp/php-cs-fixer": "^3.75",
+        "phpunit/phpunit": "^11",
+        "portphp/portphp": "^1.9",
+        "rector/rector": "^2.0",
+        "symfony/messenger": "^7.0"
    },
    "suggest": {
        "amphp/amp": "Provide asynchronous steps for your dataflows",
--- a/phpunit.xml
+++ b/phpunit.xml
@@ -1,15 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!-- http://www.phpunit.de/manual/current/en/appendixes.configuration.html -->
-<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" backupGlobals="false" backupStaticAttributes="false" bootstrap="Tests/bootstrap.php" convertErrorsToExceptions="true" convertNoticesToExceptions="true" convertWarningsToExceptions="true" colors="false" xsi:noNamespaceSchemaLocation="https://schema.phpunit.de/9.3/phpunit.xsd">
-  <coverage>
-    <include>
-      <directory>./src/</directory>
-    </include>
-    <exclude>
-      <directory>Tests/</directory>
-      <directory>vendor/</directory>
-    </exclude>
-  </coverage>
+<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" backupGlobals="false" bootstrap="Tests/bootstrap.php" colors="false" xsi:noNamespaceSchemaLocation="https://schema.phpunit.de/11.1/phpunit.xsd" cacheDirectory=".phpunit.cache" backupStaticProperties="false">
  <php>
    <ini name="error_reporting" value="-1"/>
  </php>
@@ -18,4 +9,13 @@
      <directory suffix="Test.php">./Tests</directory>
    </testsuite>
  </testsuites>
+  <source>
+    <include>
+      <directory>./src/</directory>
+    </include>
+    <exclude>
+      <directory>Tests/</directory>
+      <directory>vendor/</directory>
+    </exclude>
+  </source>
 </phpunit>
--- a/rector.php
+++ b/rector.php
@@ -17,7 +17,7 @@ return static function (RectorConfig $rectorConfig): void {
    $rectorConfig->rule(InlineConstructorDefaultToPropertyRector::class);

    $rectorConfig->sets([
-        SymfonySetList::SYMFONY_60,
+        SymfonySetList::SYMFONY_70,
        SymfonySetList::SYMFONY_CODE_QUALITY,
        SymfonySetList::SYMFONY_CONSTRUCTOR_INJECTION,
        LevelSetList::UP_TO_PHP_80,
--- a/sonar-project.properties
+++ b/sonar-project.properties
@@ -0,0 +1,4 @@
+sonar.projectKey=code-rhapsodie_dataflow-bundle_AYvYuaAwWE9sbcQmD1vw
+
+sonar.sources=src
+sonar.tests=Tests
--- a/src/CodeRhapsodieDataflowBundle.php
+++ b/src/CodeRhapsodieDataflowBundle.php
@@ -17,14 +17,14 @@ use Symfony\Component\HttpKernel\Bundle\Bundle;
 */
 class CodeRhapsodieDataflowBundle extends Bundle
 {
-    protected $name = 'CodeRhapsodieDataflowBundle';
+    protected string $name = 'CodeRhapsodieDataflowBundle';

    public function getContainerExtension(): ?ExtensionInterface
    {
        return new CodeRhapsodieDataflowExtension();
    }

-    public function build(ContainerBuilder $container)
+    public function build(ContainerBuilder $container): void
    {
        $container
            ->addCompilerPass(new DataflowTypeCompilerPass())
--- a/src/Command/AddScheduledDataflowCommand.php
+++ b/src/Command/AddScheduledDataflowCommand.php
@@ -8,6 +8,7 @@ use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
 use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
+use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
@@ -18,39 +19,35 @@ use Symfony\Component\Validator\Validator\ValidatorInterface;
 /**
 * @codeCoverageIgnore
 */
+#[AsCommand('code-rhapsodie:dataflow:schedule:add', 'Create a scheduled dataflow')]
 class AddScheduledDataflowCommand extends Command
 {
-    protected static $defaultName = 'code-rhapsodie:dataflow:schedule:add';
-
    public function __construct(private DataflowTypeRegistryInterface $registry, private ScheduledDataflowRepository $scheduledDataflowRepository, private ValidatorInterface $validator, private ConnectionFactory $connectionFactory)
    {
        parent::__construct();
    }

-    /**
-     * {@inheritdoc}
-     */
-    protected function configure()
+    protected function configure(): void
    {
        $this
-            ->setDescription('Create a scheduled dataflow')
            ->setHelp('The <info>%command.name%</info> allows you to create a new scheduled dataflow.')
            ->addOption('label', null, InputOption::VALUE_REQUIRED, 'Label of the scheduled dataflow')
            ->addOption('type', null, InputOption::VALUE_REQUIRED, 'Type of the scheduled dataflow (FQCN)')
-            ->addOption('options', null, InputOption::VALUE_OPTIONAL,
-                'Options of the scheduled dataflow (ex: {"option1": "value1", "option2": "value2"})')
+            ->addOption(
+                'options',
+                null,
+                InputOption::VALUE_OPTIONAL,
+                'Options of the scheduled dataflow (ex: {"option1": "value1", "option2": "value2"})'
+            )
            ->addOption('frequency', null, InputOption::VALUE_REQUIRED, 'Frequency of the scheduled dataflow')
            ->addOption('first_run', null, InputOption::VALUE_REQUIRED, 'Date for the first run of the scheduled dataflow (Y-m-d H:i:s)')
            ->addOption('enabled', null, InputOption::VALUE_REQUIRED, 'State of the scheduled dataflow')
            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
    }

-    /**
-     * {@inheritdoc}
-     */
    protected function execute(InputInterface $input, OutputInterface $output): int
    {
-        if (null !== $input->getOption('connection')) {
+        if ($input->getOption('connection') !== null) {
            $this->connectionFactory->setConnectionName($input->getOption('connection'));
        }
        $choices = [];
@@ -72,13 +69,17 @@ class AddScheduledDataflowCommand extends Command
        }
        $options = $input->getOption('options');
        if (!$options) {
-            $options = $io->ask('What are the launch options for the scheduled dataflow? (ex: {"option1": "value1", "option2": "value2"})',
-                json_encode([]));
+            $options = $io->ask(
+                'What are the launch options for the scheduled dataflow? (ex: {"option1": "value1", "option2": "value2"})',
+                json_encode([])
+            );
        }
        $frequency = $input->getOption('frequency');
        if (!$frequency) {
-            $frequency = $io->choice('What is the frequency for the scheduled dataflow?',
-                ScheduledDataflow::AVAILABLE_FREQUENCIES);
+            $frequency = $io->choice(
+                'What is the frequency for the scheduled dataflow?',
+                ScheduledDataflow::AVAILABLE_FREQUENCIES
+            );
        }
        $firstRun = $input->getOption('first_run');
        if (!$firstRun) {
@@ -93,22 +94,25 @@ class AddScheduledDataflowCommand extends Command
            'id' => null,
            'label' => $label,
            'dataflow_type' => $type,
-            'options' => json_decode($options, true, 512, JSON_THROW_ON_ERROR),
+            'options' => json_decode($options, true, 512, \JSON_THROW_ON_ERROR),
            'frequency' => $frequency,
-            'next' => new \DateTimeImmutable($firstRun),
+            'next' => new \DateTime($firstRun),
            'enabled' => $enabled,
        ]);

        $errors = $this->validator->validate($newScheduledDataflow);
-        if (count($errors) > 0) {
+        if (\count($errors) > 0) {
            $io->error((string) $errors);

            return 2;
        }

        $this->scheduledDataflowRepository->save($newScheduledDataflow);
-        $io->success(sprintf('New scheduled dataflow "%s" (id:%d) was created successfully.',
-            $newScheduledDataflow->getLabel(), $newScheduledDataflow->getId()));
+        $io->success(\sprintf(
+            'New scheduled dataflow "%s" (id:%d) was created successfully.',
+            $newScheduledDataflow->getLabel(),
+            $newScheduledDataflow->getId()
+        ));

        return 0;
    }
--- a/src/Command/ChangeScheduleStatusCommand.php
+++ b/src/Command/ChangeScheduleStatusCommand.php
@@ -7,6 +7,7 @@ namespace CodeRhapsodie\DataflowBundle\Command;
 use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
 use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
+use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
@@ -17,22 +18,17 @@ use Symfony\Component\Console\Style\SymfonyStyle;
 /**
 * @codeCoverageIgnore
 */
+#[AsCommand('code-rhapsodie:dataflow:schedule:change-status', 'Change schedule status')]
 class ChangeScheduleStatusCommand extends Command
 {
-    protected static $defaultName = 'code-rhapsodie:dataflow:schedule:change-status';
-
    public function __construct(private ScheduledDataflowRepository $scheduledDataflowRepository, private ConnectionFactory $connectionFactory)
    {
        parent::__construct();
    }

-    /**
-     * {@inheritdoc}
-     */
-    protected function configure()
+    protected function configure(): void
    {
        $this
-            ->setDescription('Change schedule status')
            ->setHelp('The <info>%command.name%</info> command able you to change schedule status.')
            ->addArgument('schedule-id', InputArgument::REQUIRED, 'Id of the schedule')
            ->addOption('enable', null, InputOption::VALUE_NONE, 'Enable the schedule')
@@ -40,12 +36,9 @@ class ChangeScheduleStatusCommand extends Command
            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
    }

-    /**
-     * {@inheritdoc}
-     */
    protected function execute(InputInterface $input, OutputInterface $output): int
    {
-        if (null !== $input->getOption('connection')) {
+        if ($input->getOption('connection') !== null) {
            $this->connectionFactory->setConnectionName($input->getOption('connection'));
        }
        $io = new SymfonyStyle($input, $output);
@@ -53,7 +46,7 @@ class ChangeScheduleStatusCommand extends Command
        $schedule = $this->scheduledDataflowRepository->find((int) $input->getArgument('schedule-id'));

        if (!$schedule) {
-            $io->error(sprintf('Cannot find scheduled dataflow with id "%d".', $input->getArgument('schedule-id')));
+            $io->error(\sprintf('Cannot find scheduled dataflow with id "%d".', $input->getArgument('schedule-id')));

            return 1;
        }
@@ -72,9 +65,9 @@ class ChangeScheduleStatusCommand extends Command
        try {
            $schedule->setEnabled($input->getOption('enable'));
            $this->scheduledDataflowRepository->save($schedule);
-            $io->success(sprintf('Schedule with id "%s" has been successfully updated.', $schedule->getId()));
+            $io->success(\sprintf('Schedule with id "%s" has been successfully updated.', $schedule->getId()));
        } catch (\Exception $e) {
-            $io->error(sprintf('An error occured when changing schedule status : "%s".', $e->getMessage()));
+            $io->error(\sprintf('An error occured when changing schedule status : "%s".', $e->getMessage()));

            return 4;
        }
--- a/src/Command/DatabaseSchemaCommand.php
+++ b/src/Command/DatabaseSchemaCommand.php
@@ -0,0 +1,110 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\Command;
+
+use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
+use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
+use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
+use CodeRhapsodie\DataflowBundle\SchemaProvider\DataflowSchemaProvider;
+use Doctrine\DBAL\Schema\Comparator;
+use Doctrine\DBAL\Schema\Schema;
+use Doctrine\DBAL\Schema\Table;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Question\ConfirmationQuestion;
+use Symfony\Component\Console\Style\SymfonyStyle;
+
+#[AsCommand(name: 'code-rhapsodie:dataflow:database-schema', description: 'Generates schema create / update SQL queries')]
+class DatabaseSchemaCommand extends Command
+{
+    public function __construct(private ConnectionFactory $connectionFactory)
+    {
+        parent::__construct();
+    }
+
+    protected function configure(): void
+    {
+        $this
+            ->setHelp('The <info>%command.name%</info> help you to generate SQL Query to create or update your database schema for this bundle')
+            ->addOption('dump-sql', null, InputOption::VALUE_NONE, 'Dump only the update SQL queries.')
+            ->addOption('update', null, InputOption::VALUE_NONE, 'Dump/execute only the update SQL queries.')
+            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $io = new SymfonyStyle($input, $output);
+
+        if ($input->getOption('connection') !== null) {
+            $this->connectionFactory->setConnectionName($input->getOption('connection'));
+        }
+
+        $connection = $this->connectionFactory->getConnection();
+
+        $schemaProvider = new DataflowSchemaProvider();
+        $schema = $schemaProvider->createSchema();
+
+        $sqls = $schema->toSql($connection->getDatabasePlatform());
+
+        if ($input->getOption('update')) {
+            $sm = $connection->createSchemaManager();
+
+            $tableArray = [JobRepository::TABLE_NAME, ScheduledDataflowRepository::TABLE_NAME];
+            $tables = [];
+            foreach ($sm->listTables() as $table) {
+                /** @var Table $table */
+                if (\in_array($table->getName(), $tableArray)) {
+                    $tables[] = $table;
+                }
+            }
+
+            $namespaces = [];
+
+            if ($connection->getDatabasePlatform()->supportsSchemas()) {
+                $namespaces = $sm->listSchemaNames();
+            }
+
+            $sequences = [];
+
+            if ($connection->getDatabasePlatform()->supportsSequences()) {
+                $sequences = $sm->listSequences();
+            }
+
+            $oldSchema = new Schema($tables, $sequences, $sm->createSchemaConfig(), $namespaces);
+
+            $sqls = $connection->getDatabasePlatform()->getAlterSchemaSQL((new Comparator($connection->getDatabasePlatform()))->compareSchemas($oldSchema, $schema));
+
+            if (empty($sqls)) {
+                $io->info('There is no update SQL queries.');
+            }
+        }
+
+        if ($input->getOption('dump-sql')) {
+            $io->text('Execute these SQL Queries on your database:');
+            foreach ($sqls as $sql) {
+                $io->text($sql.';');
+            }
+
+            return Command::SUCCESS;
+        }
+
+        if (!$io->askQuestion(new ConfirmationQuestion('Are you sure to update database ?', true))) {
+            $io->text('Execution canceled.');
+
+            return Command::SUCCESS;
+        }
+
+        foreach ($sqls as $sql) {
+            $connection->executeQuery($sql);
+        }
+
+        $io->success(\sprintf('%d queries executed.', \count($sqls)));
+
+        return parent::SUCCESS;
+    }
+}
--- a/src/Command/ExecuteDataflowCommand.php
+++ b/src/Command/ExecuteDataflowCommand.php
@@ -4,10 +4,13 @@ declare(strict_types=1);

 namespace CodeRhapsodie\DataflowBundle\Command;

+use CodeRhapsodie\DataflowBundle\DataflowType\AutoUpdateCountInterface;
 use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
+use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
 use Psr\Log\LoggerAwareInterface;
 use Psr\Log\LoggerAwareTrait;
+use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
@@ -20,25 +23,21 @@ use Symfony\Component\Console\Style\SymfonyStyle;
 *
 * @codeCoverageIgnore
 */
+#[AsCommand('code-rhapsodie:dataflow:execute', 'Runs one dataflow type with provided options')]
 class ExecuteDataflowCommand extends Command implements LoggerAwareInterface
 {
    use LoggerAwareTrait;

-    protected static $defaultName = 'code-rhapsodie:dataflow:execute';
-
-    public function __construct(private DataflowTypeRegistryInterface $registry, private ConnectionFactory $connectionFactory)
+    public function __construct(private DataflowTypeRegistryInterface $registry, private ConnectionFactory $connectionFactory, private JobRepository $jobRepository)
    {
        parent::__construct();
    }

-    /**
-     * {@inheritdoc}
-     */
-    protected function configure()
+    protected function configure(): void
    {
        $this
-            ->setDescription('Runs one dataflow type with provided options')
-            ->setHelp(<<<'EOF'
+            ->setHelp(
+                <<<'EOF'
 The <info>%command.name%</info> command runs one dataflow with the provided options.

  <info>php %command.full_name% App\Dataflow\MyDataflow '{"option1": "value1", "option2": "value2"}'</info>
@@ -49,19 +48,20 @@ EOF
            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
    }

-    /**
-     * {@inheritdoc}
-     */
    protected function execute(InputInterface $input, OutputInterface $output): int
    {
-        if (null !== $input->getOption('connection')) {
+        if ($input->getOption('connection') !== null) {
            $this->connectionFactory->setConnectionName($input->getOption('connection'));
        }
        $fqcnOrAlias = $input->getArgument('fqcn');
-        $options = json_decode($input->getArgument('options'), true, 512, JSON_THROW_ON_ERROR);
+        $options = json_decode($input->getArgument('options'), true, 512, \JSON_THROW_ON_ERROR);
        $io = new SymfonyStyle($input, $output);

        $dataflowType = $this->registry->getDataflowType($fqcnOrAlias);
+        if ($dataflowType instanceof AutoUpdateCountInterface) {
+            $dataflowType->setRepository($this->jobRepository);
+        }
+
        if ($dataflowType instanceof LoggerAwareInterface && isset($this->logger)) {
            $dataflowType->setLogger($this->logger);
        }
--- a/src/Command/JobShowCommand.php
+++ b/src/Command/JobShowCommand.php
@@ -7,6 +7,7 @@ namespace CodeRhapsodie\DataflowBundle\Command;
 use CodeRhapsodie\DataflowBundle\Entity\Job;
 use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
+use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
@@ -16,6 +17,7 @@ use Symfony\Component\Console\Style\SymfonyStyle;
 /**
 * @codeCoverageIgnore
 */
+#[AsCommand('code-rhapsodie:dataflow:job:show', 'Display job details for schedule or specific job')]
 class JobShowCommand extends Command
 {
    private const STATUS_MAPPING = [
@@ -24,20 +26,14 @@ class JobShowCommand extends Command
        Job::STATUS_COMPLETED => 'Completed',
    ];

-    protected static $defaultName = 'code-rhapsodie:dataflow:job:show';
-
    public function __construct(private JobRepository $jobRepository, private ConnectionFactory $connectionFactory)
    {
        parent::__construct();
    }

-    /**
-     * {@inheritdoc}
-     */
-    protected function configure()
+    protected function configure(): void
    {
        $this
-            ->setDescription('Display job details for schedule or specific job')
            ->setHelp('The <info>%command.name%</info> display job details for schedule or specific job.')
            ->addOption('job-id', null, InputOption::VALUE_REQUIRED, 'Id of the job to get details')
            ->addOption('schedule-id', null, InputOption::VALUE_REQUIRED, 'Id of schedule for last execution details')
@@ -45,12 +41,9 @@ class JobShowCommand extends Command
            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
    }

-    /**
-     * {@inheritdoc}
-     */
    protected function execute(InputInterface $input, OutputInterface $output): int
    {
-        if (null !== $input->getOption('connection')) {
+        if ($input->getOption('connection') !== null) {
            $this->connectionFactory->setConnectionName($input->getOption('connection'));
        }

@@ -74,7 +67,7 @@ class JobShowCommand extends Command
            return 2;
        }

-        if (null === $job) {
+        if ($job === null) {
            $io->error('Cannot find job :/');

            return 3;
@@ -88,19 +81,19 @@ class JobShowCommand extends Command
            ['Started at', $job->getStartTime() ? $job->getStartTime()->format('Y-m-d H:i:s') : '-'],
            ['Ended at', $job->getEndTime() ? $job->getEndTime()->format('Y-m-d H:i:s') : '-'],
            ['Object number', $job->getCount()],
-            ['Errors', count((array) $job->getExceptions())],
+            ['Errors', \count((array) $job->getExceptions())],
            ['Status', $this->translateStatus($job->getStatus())],
        ];
        if ($input->getOption('details')) {
            $display[] = ['Type', $job->getDataflowType()];
-            $display[] = ['Options', json_encode($job->getOptions(), JSON_THROW_ON_ERROR)];
+            $display[] = ['Options', json_encode($job->getOptions(), \JSON_THROW_ON_ERROR)];
            $io->section('Summary');
        }

        $io->table(['Field', 'Value'], $display);
        if ($input->getOption('details')) {
            $io->section('Exceptions');
-            $exceptions = array_map(fn(string $exception) => substr($exception, 0, 900).'…', $job->getExceptions());
+            $exceptions = array_map(fn (string $exception) => substr($exception, 0, 900).'…', $job->getExceptions());

            $io->write($exceptions);
        }
--- a/src/Command/RunPendingDataflowsCommand.php
+++ b/src/Command/RunPendingDataflowsCommand.php
@@ -7,6 +7,7 @@ namespace CodeRhapsodie\DataflowBundle\Command;
 use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Manager\ScheduledDataflowManagerInterface;
 use CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunnerInterface;
+use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Command\LockableTrait;
 use Symfony\Component\Console\Input\InputInterface;
@@ -18,34 +19,27 @@ use Symfony\Component\Console\Output\OutputInterface;
 *
 * @codeCoverageIgnore
 */
+#[AsCommand('code-rhapsodie:dataflow:run-pending', 'Runs dataflows based on the scheduled defined in the UI.')]
 class RunPendingDataflowsCommand extends Command
 {
    use LockableTrait;

-    protected static $defaultName = 'code-rhapsodie:dataflow:run-pending';
-
    public function __construct(private ScheduledDataflowManagerInterface $manager, private PendingDataflowRunnerInterface $runner, private ConnectionFactory $connectionFactory)
    {
        parent::__construct();
    }

-    /**
-     * {@inheritdoc}
-     */
-    protected function configure()
+    protected function configure(): void
    {
        $this
-            ->setDescription('Runs dataflows based on the scheduled defined in the UI.')
-            ->setHelp(<<<'EOF'
+            ->setHelp(
+                <<<'EOF'
 The <info>%command.name%</info> command runs dataflows according to the schedule defined in the UI by the user.
 EOF
            )
            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
    }

-    /**
-     * {@inheritdoc}
-     */
    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        if (!$this->lock()) {
@@ -54,7 +48,7 @@ EOF
            return 0;
        }

-        if (null !== $input->getOption('connection')) {
+        if ($input->getOption('connection') !== null) {
            $this->connectionFactory->setConnectionName($input->getOption('connection'));
        }

--- a/src/Command/ScheduleListCommand.php
+++ b/src/Command/ScheduleListCommand.php
@@ -6,6 +6,7 @@ namespace CodeRhapsodie\DataflowBundle\Command;

 use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
+use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
@@ -15,32 +16,24 @@ use Symfony\Component\Console\Style\SymfonyStyle;
 /**
 * @codeCoverageIgnore
 */
+#[AsCommand('code-rhapsodie:dataflow:schedule:list', 'List scheduled dataflows')]
 class ScheduleListCommand extends Command
 {
-    protected static $defaultName = 'code-rhapsodie:dataflow:schedule:list';
-
    public function __construct(private ScheduledDataflowRepository $scheduledDataflowRepository, private ConnectionFactory $connectionFactory)
    {
        parent::__construct();
    }

-    /**
-     * {@inheritdoc}
-     */
-    protected function configure()
+    protected function configure(): void
    {
        $this
-            ->setDescription('List scheduled dataflows')
            ->setHelp('The <info>%command.name%</info> lists all scheduled dataflows.')
            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
    }

-    /**
-     * {@inheritdoc}
-     */
    protected function execute(InputInterface $input, OutputInterface $output): int
    {
-        if (null !== $input->getOption('connection')) {
+        if ($input->getOption('connection') !== null) {
            $this->connectionFactory->setConnectionName($input->getOption('connection'));
        }
        $io = new SymfonyStyle($input, $output);
--- a/src/Command/SchemaCommand.php
+++ b/src/Command/SchemaCommand.php
@@ -4,13 +4,9 @@ declare(strict_types=1);

 namespace CodeRhapsodie\DataflowBundle\Command;

-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
-use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
-use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
-use CodeRhapsodie\DataflowBundle\SchemaProvider\DataflowSchemaProvider;
-use Doctrine\DBAL\Schema\Schema;
-use Doctrine\DBAL\Schema\Table;
+use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\ArrayInput;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
@@ -18,79 +14,41 @@ use Symfony\Component\Console\Style\SymfonyStyle;

 /**
 * @codeCoverageIgnore
+ *
+ * @deprecated This command is deprecated and will be removed in 6.0, use this command "code-rhapsodie:dataflow:database-schema" instead.
 */
+#[AsCommand('code-rhapsodie:dataflow:dump-schema', 'Generates schema create / update SQL queries')]
 class SchemaCommand extends Command
 {
-    protected static $defaultName = 'code-rhapsodie:dataflow:dump-schema';
-
-    public function __construct(private ConnectionFactory $connectionFactory)
-    {
-        parent::__construct();
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    protected function configure()
+    protected function configure(): void
    {
        $this
-            ->setDescription('Generates schema create / update SQL queries')
            ->setHelp('The <info>%command.name%</info> help you to generate SQL Query to create or update your database schema for this bundle')
            ->addOption('update', null, InputOption::VALUE_NONE, 'Dump only the update SQL queries.')
            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use')
        ;
    }

-    /**
-     * {@inheritdoc}
-     */
    protected function execute(InputInterface $input, OutputInterface $output): int
    {
-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
-
-        $connection = $this->connectionFactory->getConnection();
-
-        $schemaProvider = new DataflowSchemaProvider();
-        $schema = $schemaProvider->createSchema();
-
-        $sqls = $schema->toSql($connection->getDatabasePlatform());
-
-        if ($input->getOption('update')) {
-            $sm = $connection->getSchemaManager();
-
-            $tableArray = [JobRepository::TABLE_NAME, ScheduledDataflowRepository::TABLE_NAME];
-            $tables = [];
-            foreach ($sm->listTables() as $table) {
-                /** @var Table $table */
-                if (in_array($table->getName(), $tableArray)) {
-                    $tables[] = $table;
-                }
-            }
-
-            $namespaces = [];
-
-            if ($connection->getDatabasePlatform()->supportsSchemas()) {
-                $namespaces = $sm->listNamespaceNames();
-            }
-
-            $sequences = [];
-
-            if ($connection->getDatabasePlatform()->supportsSequences()) {
-                $sequences = $sm->listSequences();
-            }
-
-            $oldSchema = new Schema($tables, $sequences, $sm->createSchemaConfig(), $namespaces);
-
-            $sqls = $schema->getMigrateFromSql($oldSchema, $connection->getDatabasePlatform());
-        }
        $io = new SymfonyStyle($input, $output);
-        $io->text('Execute these SQL Queries on your database:');
-        foreach ($sqls as $sql) {
-            $io->text($sql.';');
-        }
+        $io->warning('This command is deprecated and will be removed in 6.0, use this command "code-rhapsodie:dataflow:database-schema" instead.');

-        return 0;
+        $options = array_filter($input->getOptions());
+
+        // add -- before each keys
+        $options = array_combine(
+            array_map(fn ($key) => '--'.$key, array_keys($options)),
+            array_values($options)
+        );
+
+        $options['--dump-sql'] = true;
+
+        $inputArray = new ArrayInput([
+            'command' => 'code-rhapsodie:dataflow:database-schema',
+            ...$options,
+        ]);
+
+        return $this->getApplication()->doRun($inputArray, $output);
    }
 }
--- a/src/DataflowType/AMPAsyncDataflowBuilder.php
+++ b/src/DataflowType/AMPAsyncDataflowBuilder.php
@@ -7,23 +7,20 @@ namespace CodeRhapsodie\DataflowBundle\DataflowType;
 use CodeRhapsodie\DataflowBundle\DataflowType\Dataflow\AMPAsyncDataflow;
 use CodeRhapsodie\DataflowBundle\DataflowType\Dataflow\DataflowInterface;
 use CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface;
-use Symfony\Component\OptionsResolver\OptionsResolver;

 class AMPAsyncDataflowBuilder extends DataflowBuilder
 {
-    public function __construct(protected ?int $loopInterval = 0, protected ?int $emitInterval = 0)
-    {
-    }
-
    private ?string $name = null;
-
    private ?iterable $reader = null;
-
    private array $steps = [];

    /** @var WriterInterface[] */
    private array $writers = [];

+    public function __construct(protected ?int $loopInterval = 0, protected ?int $emitInterval = 0)
+    {
+    }
+
    public function setName(string $name): self
    {
        $this->name = $name;
--- a/src/DataflowType/AbstractDataflowType.php
+++ b/src/DataflowType/AbstractDataflowType.php
@@ -4,15 +4,20 @@ declare(strict_types=1);

 namespace CodeRhapsodie\DataflowBundle\DataflowType;

+use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
 use Psr\Log\LoggerAwareInterface;
 use Psr\Log\LoggerAwareTrait;
 use Psr\Log\LoggerInterface;
 use Symfony\Component\OptionsResolver\OptionsResolver;

-abstract class AbstractDataflowType implements DataflowTypeInterface, LoggerAwareInterface
+abstract class AbstractDataflowType implements DataflowTypeInterface, LoggerAwareInterface, AutoUpdateCountInterface
 {
    use LoggerAwareTrait;

+    private JobRepository $repository;
+
+    private ?\DateTime $saveDate = null;
+
    /**
     * @codeCoverageIgnore
     */
@@ -21,14 +26,24 @@ abstract class AbstractDataflowType implements DataflowTypeInterface, LoggerAwar
        return [];
    }

-    public function process(array $options): Result
+    public function process(array $options, ?int $jobId = null): Result
    {
+        $this->saveDate = new \DateTime('+1 minute');
+
        $optionsResolver = new OptionsResolver();
        $this->configureOptions($optionsResolver);
        $options = $optionsResolver->resolve($options);

        $builder = $this->createDataflowBuilder();
        $builder->setName($this->getLabel());
+        $builder->addAfterItemProcessor(function (int|string $index, mixed $item, int $count) use ($jobId) {
+            if ($jobId === null || $this->saveDate > new \DateTime()) {
+                return;
+            }
+
+            $this->repository->updateCount($jobId, $count);
+            $this->saveDate = new \DateTime('+1 minute');
+        });
        $this->buildDataflow($builder, $options);
        $dataflow = $builder->getDataflow();
        if ($dataflow instanceof LoggerAwareInterface && $this->logger instanceof LoggerInterface) {
@@ -38,6 +53,11 @@ abstract class AbstractDataflowType implements DataflowTypeInterface, LoggerAwar
        return $dataflow->process();
    }

+    public function setRepository(JobRepository $repository): void
+    {
+        $this->repository = $repository;
+    }
+
    protected function createDataflowBuilder(): DataflowBuilder
    {
        return new DataflowBuilder();
--- a/src/DataflowType/AutoUpdateCountInterface.php
+++ b/src/DataflowType/AutoUpdateCountInterface.php
@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\DataflowType;
+
+use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
+
+interface AutoUpdateCountInterface
+{
+    public function setRepository(JobRepository $repository): void;
+}
--- a/src/DataflowType/Dataflow/AMPAsyncDataflow.php
+++ b/src/DataflowType/Dataflow/AMPAsyncDataflow.php
@@ -5,18 +5,19 @@ declare(strict_types=1);
 namespace CodeRhapsodie\DataflowBundle\DataflowType\Dataflow;

 use function Amp\coroutine;
+
 use Amp\Deferred;
 use Amp\Delayed;
 use Amp\Loop;
 use Amp\Producer;
 use Amp\Promise;
+
 use function Amp\Promise\wait;
+
 use CodeRhapsodie\DataflowBundle\DataflowType\Result;
 use CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface;
 use Psr\Log\LoggerAwareInterface;
 use Psr\Log\LoggerAwareTrait;
-use RuntimeException;
-use Throwable;

 class AMPAsyncDataflow implements DataflowInterface, LoggerAwareInterface
 {
@@ -34,8 +35,8 @@ class AMPAsyncDataflow implements DataflowInterface, LoggerAwareInterface

    public function __construct(private iterable $reader, private ?string $name, private ?int $loopInterval = 0, private ?int $emitInterval = 0)
    {
-        if (!function_exists('Amp\\Promise\\wait')) {
-            throw new RuntimeException('Amp is not loaded. Suggest install it with composer require amphp/amp');
+        if (!\function_exists('Amp\\Promise\\wait')) {
+            throw new \RuntimeException('Amp is not loaded. Suggest install it with composer require amphp/amp');
        }
    }

@@ -61,14 +62,11 @@ class AMPAsyncDataflow implements DataflowInterface, LoggerAwareInterface
        return $this;
    }

-    /**
-     * {@inheritdoc}
-     */
    public function process(): Result
    {
        $count = 0;
        $exceptions = [];
-        $startTime = new \DateTimeImmutable();
+        $startTime = new \DateTime();

        try {
            foreach ($this->writers as $writer) {
@@ -76,7 +74,7 @@ class AMPAsyncDataflow implements DataflowInterface, LoggerAwareInterface
            }

            $deferred = new Deferred();
-            $resolved = false; //missing $deferred->isResolved() in version 2.5
+            $resolved = false; // missing $deferred->isResolved() in version 2.5
            $producer = new Producer(function (callable $emit) {
                foreach ($this->reader as $index => $item) {
                    yield new Delayed($this->emitInterval);
@@ -89,7 +87,7 @@ class AMPAsyncDataflow implements DataflowInterface, LoggerAwareInterface
                    $it = $producer->getCurrent();
                    [$index, $item] = $it;
                    $this->states[$index] = [$index, 0, $item];
-                } elseif (!$resolved && 0 === count($this->states)) {
+                } elseif (!$resolved && \count($this->states) === 0) {
                    $resolved = true;
                    $deferred->resolve();
                }
@@ -110,7 +108,7 @@ class AMPAsyncDataflow implements DataflowInterface, LoggerAwareInterface
            $this->logException($e);
        }

-        return new Result($this->name, $startTime, new \DateTimeImmutable(), $count, $exceptions);
+        return new Result($this->name, $startTime, new \DateTime(), $count, $exceptions);
    }

    /**
@@ -120,20 +118,20 @@ class AMPAsyncDataflow implements DataflowInterface, LoggerAwareInterface
    private function processState(mixed $state, int &$count, array &$exceptions): void
    {
        [$readIndex, $stepIndex, $item] = $state;
-        if ($stepIndex < count($this->steps)) {
+        if ($stepIndex < \count($this->steps)) {
            if (!isset($this->stepsJobs[$stepIndex])) {
                $this->stepsJobs[$stepIndex] = [];
            }
            [$step, $scale] = $this->steps[$stepIndex];
-            if ((is_countable($this->stepsJobs[$stepIndex]) ? count($this->stepsJobs[$stepIndex]) : 0) < $scale && !isset($this->stepsJobs[$stepIndex][$readIndex])) {
+            if ((is_countable($this->stepsJobs[$stepIndex]) ? \count($this->stepsJobs[$stepIndex]) : 0) < $scale && !isset($this->stepsJobs[$stepIndex][$readIndex])) {
                $this->stepsJobs[$stepIndex][$readIndex] = true;
                /** @var Promise<void> $promise */
                $promise = coroutine($step)($item);
-                $promise->onResolve(function (?Throwable $exception = null, $newItem = null) use ($stepIndex, $readIndex, &$exceptions) {
+                $promise->onResolve(function (?\Throwable $exception = null, $newItem = null) use ($stepIndex, $readIndex, &$exceptions) {
                    if ($exception) {
                        $exceptions[$stepIndex] = $exception;
                        $this->logException($exception, (string) $stepIndex);
-                    } elseif (false === $newItem) {
+                    } elseif ($newItem === false) {
                        unset($this->states[$readIndex]);
                    } else {
                        $this->states[$readIndex] = [$readIndex, $stepIndex + 1, $newItem];
@@ -153,7 +151,7 @@ class AMPAsyncDataflow implements DataflowInterface, LoggerAwareInterface
        }
    }

-    private function logException(Throwable $e, ?string $index = null): void
+    private function logException(\Throwable $e, ?string $index = null): void
    {
        if (!isset($this->logger)) {
            return;
--- a/src/DataflowType/Dataflow/Dataflow.php
+++ b/src/DataflowType/Dataflow/Dataflow.php
@@ -21,6 +21,13 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface

    private ?\Closure $customExceptionIndex = null;

+    private ?\DateTimeInterface $dateTime = null;
+
+    /**
+     * @var \Closure[]
+     */
+    private array $afterItemProcessors = [];
+
    public function __construct(private iterable $reader, private ?string $name)
    {
    }
@@ -56,13 +63,20 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface
    }

    /**
-     * {@inheritdoc}
+     * @param array<callable> $processors
     */
+    public function setAfterItemProcessors(array $processors): self
+    {
+        $this->afterItemProcessors = array_map(fn (callable $callable) => \Closure::fromCallable($callable), $processors);
+
+        return $this;
+    }
+
    public function process(): Result
    {
        $count = 0;
        $exceptions = [];
-        $startTime = new \DateTimeImmutable();
+        $startTime = new \DateTime();

        try {
            foreach ($this->writers as $writer) {
@@ -75,7 +89,7 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface
                } catch (\Throwable $e) {
                    $exceptionIndex = $index;
                    try {
-                        if (is_callable($this->customExceptionIndex)) {
+                        if (\is_callable($this->customExceptionIndex)) {
                            $exceptionIndex = (string) ($this->customExceptionIndex)($item, $index);
                        }
                    } catch (\Throwable $e2) {
@@ -87,6 +101,10 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface
                }

                ++$count;
+
+                foreach ($this->afterItemProcessors as $afterItemProcessor) {
+                    $afterItemProcessor($index, $item, $count);
+                }
            }

            foreach ($this->writers as $writer) {
@@ -97,15 +115,15 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface
            $this->logException($e);
        }

-        return new Result($this->name, $startTime, new \DateTimeImmutable(), $count, $exceptions);
+        return new Result($this->name, $startTime, new \DateTime(), $count, $exceptions);
    }

    private function processItem(mixed $item): void
    {
        foreach ($this->steps as $step) {
-            $item = call_user_func($step, $item);
+            $item = \call_user_func($step, $item);

-            if (false === $item) {
+            if ($item === false) {
                return;
            }
        }
@@ -115,7 +133,7 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface
        }
    }

-    private function logException(\Throwable $e, ?string $index = null): void
+    private function logException(\Throwable $e, string|int|null $index = null): void
    {
        if (!isset($this->logger)) {
            return;
--- a/src/DataflowType/DataflowBuilder.php
+++ b/src/DataflowType/DataflowBuilder.php
@@ -11,15 +11,17 @@ use CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface;
 class DataflowBuilder
 {
    private ?string $name = null;
-
    private ?iterable $reader = null;
-
    private array $steps = [];

    /** @var WriterInterface[] */
    private array $writers = [];

    private ?\Closure $customExceptionIndex = null;
+    /**
+     * @var \Closure[]
+     */
+    private array $afterItemProcessors = [];

    public function setName(string $name): self
    {
@@ -56,6 +58,13 @@ class DataflowBuilder
        return $this;
    }

+    public function addAfterItemProcessor(callable $callable): self
+    {
+        $this->afterItemProcessors[] = \Closure::fromCallable($callable);
+
+        return $this;
+    }
+
    public function getDataflow(): DataflowInterface
    {
        $dataflow = new Dataflow($this->reader, $this->name);
@@ -71,10 +80,12 @@ class DataflowBuilder
            $dataflow->addWriter($writer);
        }

-        if (is_callable($this->customExceptionIndex)) {
+        if (\is_callable($this->customExceptionIndex)) {
            $dataflow->setCustomExceptionIndex($this->customExceptionIndex);
        }

+        $dataflow->setAfterItemProcessors($this->afterItemProcessors);
+
        return $dataflow;
    }
 }
--- a/src/DataflowType/DataflowTypeInterface.php
+++ b/src/DataflowType/DataflowTypeInterface.php
@@ -10,5 +10,5 @@ interface DataflowTypeInterface

    public function getAliases(): iterable;

-    public function process(array $options): Result;
+    public function process(array $options, ?int $jobId = null): Result;
 }
--- a/src/DataflowType/Result.php
+++ b/src/DataflowType/Result.php
@@ -20,7 +20,7 @@ class Result
    public function __construct(private string $name, private \DateTimeInterface $startTime, private \DateTimeInterface $endTime, private int $totalProcessedCount, array $exceptions)
    {
        $this->elapsed = $startTime->diff($endTime);
-        $this->errorCount = count($exceptions);
+        $this->errorCount = \count($exceptions);
        $this->successCount = $totalProcessedCount - $this->errorCount;
        $this->exceptions = $exceptions;
    }
--- a/src/DataflowType/Writer/CollectionWriter.php
+++ b/src/DataflowType/Writer/CollectionWriter.php
@@ -18,21 +18,15 @@ class CollectionWriter implements DelegateWriterInterface
    {
    }

-    /**
-     * {@inheritdoc}
-     */
    public function prepare()
    {
        $this->writer->prepare();
    }

-    /**
-     * {@inheritdoc}
-     */
    public function write($collection)
    {
        if (!is_iterable($collection)) {
-            throw new UnsupportedItemTypeException(sprintf('Item to write was expected to be an iterable, received %s.', get_debug_type($collection)));
+            throw new UnsupportedItemTypeException(\sprintf('Item to write was expected to be an iterable, received %s.', get_debug_type($collection)));
        }

        foreach ($collection as $item) {
@@ -40,17 +34,11 @@ class CollectionWriter implements DelegateWriterInterface
        }
    }

-    /**
-     * {@inheritdoc}
-     */
    public function finish()
    {
        $this->writer->finish();
    }

-    /**
-     * {@inheritdoc}
-     */
    public function supports($item): bool
    {
        return is_iterable($item);
--- a/src/DataflowType/Writer/DelegateWriterInterface.php
+++ b/src/DataflowType/Writer/DelegateWriterInterface.php
@@ -11,8 +11,6 @@ interface DelegateWriterInterface extends WriterInterface
 {
    /**
     * Returns true if the argument is of a supported type.
-     *
-     * @param $item
     */
    public function supports($item): bool;
 }
--- a/src/DataflowType/Writer/DelegatorWriter.php
+++ b/src/DataflowType/Writer/DelegatorWriter.php
@@ -21,9 +21,6 @@ class DelegatorWriter implements DelegateWriterInterface
    {
    }

-    /**
-     * {@inheritdoc}
-     */
    public function prepare()
    {
        foreach ($this->delegates as $delegate) {
@@ -31,9 +28,6 @@ class DelegatorWriter implements DelegateWriterInterface
        }
    }

-    /**
-     * {@inheritdoc}
-     */
    public function write($item)
    {
        foreach ($this->delegates as $delegate) {
@@ -46,12 +40,9 @@ class DelegatorWriter implements DelegateWriterInterface
            return;
        }

-        throw new UnsupportedItemTypeException(sprintf('None of the registered delegate writers support the received item of type %s', get_debug_type($item)));
+        throw new UnsupportedItemTypeException(\sprintf('None of the registered delegate writers support the received item of type %s', get_debug_type($item)));
    }

-    /**
-     * {@inheritdoc}
-     */
    public function finish()
    {
        foreach ($this->delegates as $delegate) {
@@ -59,9 +50,6 @@ class DelegatorWriter implements DelegateWriterInterface
        }
    }

-    /**
-     * {@inheritdoc}
-     */
    public function supports($item): bool
    {
        foreach ($this->delegates as $delegate) {
--- a/src/DependencyInjection/Compiler/BusCompilerPass.php
+++ b/src/DependencyInjection/Compiler/BusCompilerPass.php
@@ -12,9 +12,6 @@ use Symfony\Component\DependencyInjection\Reference;

 class BusCompilerPass implements CompilerPassInterface
 {
-    /**
-     * {@inheritdoc}
-     */
    public function process(ContainerBuilder $container)
    {
        if (!$container->hasParameter('coderhapsodie.dataflow.bus')) {
@@ -23,7 +20,7 @@ class BusCompilerPass implements CompilerPassInterface

        $bus = $container->getParameter('coderhapsodie.dataflow.bus');
        if (!$container->has($bus)) {
-            throw new InvalidArgumentException(sprintf('Service "%s" not found', $bus));
+            throw new InvalidArgumentException(\sprintf('Service "%s" not found', $bus));
        }

        if (!$container->has(MessengerDataflowRunner::class)) {
--- a/src/DependencyInjection/Compiler/DataflowTypeCompilerPass.php
+++ b/src/DependencyInjection/Compiler/DataflowTypeCompilerPass.php
@@ -16,9 +16,6 @@ use Symfony\Component\DependencyInjection\Reference;
 */
 class DataflowTypeCompilerPass implements CompilerPassInterface
 {
-    /**
-     * {@inheritdoc}
-     */
    public function process(ContainerBuilder $container)
    {
        if (!$container->has(DataflowTypeRegistry::class)) {
--- a/src/DependencyInjection/Compiler/DefaultLoggerCompilerPass.php
+++ b/src/DependencyInjection/Compiler/DefaultLoggerCompilerPass.php
@@ -12,9 +12,6 @@ use Symfony\Component\DependencyInjection\Reference;

 class DefaultLoggerCompilerPass implements CompilerPassInterface
 {
-    /**
-     * {@inheritdoc}
-     */
    public function process(ContainerBuilder $container)
    {
        $defaultLogger = $container->getParameter('coderhapsodie.dataflow.default_logger');
--- a/src/DependencyInjection/Configuration.php
+++ b/src/DependencyInjection/Configuration.php
@@ -10,15 +10,10 @@ use Symfony\Component\Messenger\MessageBusInterface;

 class Configuration implements ConfigurationInterface
 {
-    public function getConfigTreeBuilder(): \Symfony\Component\Config\Definition\Builder\TreeBuilder
+    public function getConfigTreeBuilder(): TreeBuilder
    {
        $treeBuilder = new TreeBuilder('code_rhapsodie_dataflow');
-        if (method_exists($treeBuilder, 'getRootNode')) {
-            $rootNode = $treeBuilder->getRootNode();
-        } else {
-            // BC for symfony/config < 4.2
-            $rootNode = $treeBuilder->root('code_rhapsodie_dataflow');
-        }
+        $rootNode = $treeBuilder->getRootNode();

        $rootNode
            ->children()
@@ -39,7 +34,7 @@ class Configuration implements ConfigurationInterface
                        ->end()
                    ->end()
                    ->validate()
-                        ->ifTrue(static fn($v): bool => $v['enabled'] && !interface_exists(MessageBusInterface::class))
+                        ->ifTrue(static fn ($v): bool => $v['enabled'] && !interface_exists(MessageBusInterface::class))
                        ->thenInvalid('You need "symfony/messenger" in order to use Dataflow messenger mode.')
                    ->end()
                ->end()
--- a/src/Entity/Job.php
+++ b/src/Entity/Job.php
@@ -72,26 +72,22 @@ class Job
            ->setScheduledDataflowId($scheduled->getId());
    }

-    public function __construct()
-    {
-    }
-
    public static function createFromArray(array $datas)
    {
        $lost = array_diff(static::KEYS, array_keys($datas));
-        if (count($lost) > 0) {
+        if (\count($lost) > 0) {
            throw new \LogicException('The first argument of '.__METHOD__.'  must be contains: "'.implode(', ', $lost).'"');
        }

        $job = new self();
-        $job->id = null === $datas['id'] ? null : (int) $datas['id'];
-        $job->setStatus(null === $datas['status'] ? null : (int) $datas['status']);
+        $job->id = $datas['id'] === null ? null : (int) $datas['id'];
+        $job->setStatus($datas['status'] === null ? null : (int) $datas['status']);
        $job->setLabel($datas['label']);
        $job->setDataflowType($datas['dataflow_type']);
        $job->setOptions($datas['options']);
        $job->setRequestedDate($datas['requested_date']);
-        $job->setScheduledDataflowId(null === $datas['scheduled_dataflow_id'] ? null : (int) $datas['scheduled_dataflow_id']);
-        $job->setCount(null === $datas['count'] ? null : (int) $datas['count']);
+        $job->setScheduledDataflowId($datas['scheduled_dataflow_id'] === null ? null : (int) $datas['scheduled_dataflow_id']);
+        $job->setCount($datas['count'] === null ? null : (int) $datas['count']);
        $job->setExceptions($datas['exceptions']);
        $job->setStartTime($datas['start_time']);
        $job->setEndTime($datas['end_time']);
@@ -116,7 +112,7 @@ class Job
        ];
    }

-    public function setId(int $id): Job
+    public function setId(int $id): self
    {
        $this->id = $id;

@@ -133,7 +129,7 @@ class Job
        return $this->status;
    }

-    public function setStatus(int $status): Job
+    public function setStatus(int $status): self
    {
        $this->status = $status;

@@ -145,7 +141,7 @@ class Job
        return $this->label;
    }

-    public function setLabel(?string $label): Job
+    public function setLabel(?string $label): self
    {
        $this->label = $label;

@@ -157,7 +153,7 @@ class Job
        return $this->dataflowType;
    }

-    public function setDataflowType(?string $dataflowType): Job
+    public function setDataflowType(?string $dataflowType): self
    {
        $this->dataflowType = $dataflowType;

@@ -169,7 +165,7 @@ class Job
        return $this->options;
    }

-    public function setOptions(?array $options): Job
+    public function setOptions(?array $options): self
    {
        $this->options = $options;

@@ -181,7 +177,7 @@ class Job
        return $this->requestedDate;
    }

-    public function setRequestedDate(?\DateTimeInterface $requestedDate): Job
+    public function setRequestedDate(?\DateTimeInterface $requestedDate): self
    {
        $this->requestedDate = $requestedDate;

@@ -193,7 +189,7 @@ class Job
        return $this->scheduledDataflowId;
    }

-    public function setScheduledDataflowId(?int $scheduledDataflowId): Job
+    public function setScheduledDataflowId(?int $scheduledDataflowId): self
    {
        $this->scheduledDataflowId = $scheduledDataflowId;

@@ -205,7 +201,7 @@ class Job
        return $this->count;
    }

-    public function setCount(?int $count): Job
+    public function setCount(?int $count): self
    {
        $this->count = $count;

@@ -217,7 +213,7 @@ class Job
        return $this->exceptions;
    }

-    public function setExceptions(?array $exceptions): Job
+    public function setExceptions(?array $exceptions): self
    {
        $this->exceptions = $exceptions;

@@ -229,7 +225,7 @@ class Job
        return $this->startTime;
    }

-    public function setStartTime(?\DateTimeInterface $startTime): Job
+    public function setStartTime(?\DateTimeInterface $startTime): self
    {
        $this->startTime = $startTime;

@@ -241,7 +237,7 @@ class Job
        return $this->endTime;
    }

-    public function setEndTime(?\DateTimeInterface $endTime): Job
+    public function setEndTime(?\DateTimeInterface $endTime): self
    {
        $this->endTime = $endTime;

--- a/src/Entity/ScheduledDataflow.php
+++ b/src/Entity/ScheduledDataflow.php
@@ -50,19 +50,19 @@ class ScheduledDataflow
    public static function createFromArray(array $datas)
    {
        $lost = array_diff(static::KEYS, array_keys($datas));
-        if (count($lost) > 0) {
+        if (\count($lost) > 0) {
            throw new \LogicException('The first argument of '.__METHOD__.'  must be contains: "'.implode(', ', $lost).'"');
        }

        $scheduledDataflow = new self();
-        $scheduledDataflow->id = null === $datas['id'] ? null : (int) $datas['id'];
+        $scheduledDataflow->id = $datas['id'] === null ? null : (int) $datas['id'];

        $scheduledDataflow->setLabel($datas['label']);
        $scheduledDataflow->setDataflowType($datas['dataflow_type']);
        $scheduledDataflow->setOptions($datas['options']);
        $scheduledDataflow->setFrequency($datas['frequency']);
        $scheduledDataflow->setNext($datas['next']);
-        $scheduledDataflow->setEnabled(null === $datas['enabled'] ? null : (bool) $datas['enabled']);
+        $scheduledDataflow->setEnabled($datas['enabled'] === null ? null : (bool) $datas['enabled']);

        return $scheduledDataflow;
    }
@@ -80,7 +80,7 @@ class ScheduledDataflow
        ];
    }

-    public function setId(int $id): ScheduledDataflow
+    public function setId(int $id): self
    {
        $this->id = $id;

@@ -97,7 +97,7 @@ class ScheduledDataflow
        return $this->label;
    }

-    public function setLabel(?string $label): ScheduledDataflow
+    public function setLabel(?string $label): self
    {
        $this->label = $label;

@@ -109,7 +109,7 @@ class ScheduledDataflow
        return $this->dataflowType;
    }

-    public function setDataflowType(?string $dataflowType): ScheduledDataflow
+    public function setDataflowType(?string $dataflowType): self
    {
        $this->dataflowType = $dataflowType;

@@ -121,7 +121,7 @@ class ScheduledDataflow
        return $this->options;
    }

-    public function setOptions(?array $options): ScheduledDataflow
+    public function setOptions(?array $options): self
    {
        $this->options = $options;

@@ -133,7 +133,7 @@ class ScheduledDataflow
        return $this->frequency;
    }

-    public function setFrequency(?string $frequency): ScheduledDataflow
+    public function setFrequency(?string $frequency): self
    {
        $this->frequency = $frequency;

@@ -145,7 +145,7 @@ class ScheduledDataflow
        return $this->next;
    }

-    public function setNext(?\DateTimeInterface $next): ScheduledDataflow
+    public function setNext(?\DateTimeInterface $next): self
    {
        $this->next = $next;

@@ -157,7 +157,7 @@ class ScheduledDataflow
        return $this->enabled;
    }

-    public function setEnabled(?bool $enabled): ScheduledDataflow
+    public function setEnabled(?bool $enabled): self
    {
        $this->enabled = $enabled;

--- a/src/Event/CrEvent.php
+++ b/src/Event/CrEvent.php
@@ -4,17 +4,11 @@ declare(strict_types=1);

 namespace CodeRhapsodie\DataflowBundle\Event;

-/*
+use Symfony\Contracts\EventDispatcher\Event;
+
+/**
 * @codeCoverageIgnore
 */
-if (class_exists(\Symfony\Contracts\EventDispatcher\Event::class)) {
-    // For Symfony 5.0+
-    abstract class CrEvent extends \Symfony\Contracts\EventDispatcher\Event
-    {
-    }
-} else {
-    // For Symfony 3.4 to 4.4
-    abstract class CrEvent extends \Symfony\Component\EventDispatcher\Event
-    {
-    }
+abstract class CrEvent extends Event
+{
 }
--- a/src/Exceptions/UnknownDataflowTypeException.php
+++ b/src/Exceptions/UnknownDataflowTypeException.php
@@ -11,7 +11,7 @@ class UnknownDataflowTypeException extends \Exception
 {
    public static function create(string $aliasOrFqcn, array $knownDataflowTypes): self
    {
-        return new self(sprintf(
+        return new self(\sprintf(
            'Unknown dataflow type FQCN or alias "%s". Registered dataflow types FQCN and aliases are %s.',
            $aliasOrFqcn,
            implode(', ', $knownDataflowTypes)
--- a/src/Factory/ConnectionFactory.php
+++ b/src/Factory/ConnectionFactory.php
@@ -24,6 +24,6 @@ class ConnectionFactory

    public function getConnection(): \Doctrine\DBAL\Connection
    {
-        return $this->container->get(sprintf('doctrine.dbal.%s_connection', $this->connectionName));
+        return $this->container->get(\sprintf('doctrine.dbal.%s_connection', $this->connectionName));
    }
 }
--- a/src/Logger/BufferHandler.php
+++ b/src/Logger/BufferHandler.php
@@ -7,7 +7,7 @@ namespace CodeRhapsodie\DataflowBundle\Logger;
 use Monolog\Formatter\FormatterInterface;
 use Monolog\Formatter\LineFormatter;
 use Monolog\Handler\AbstractProcessingHandler;
-use Monolog\Logger;
+use Monolog\LogRecord;

 class BufferHandler extends AbstractProcessingHandler
 {
@@ -15,11 +15,6 @@ class BufferHandler extends AbstractProcessingHandler

    private array $buffer = [];

-    public function __construct($level = Logger::DEBUG, bool $bubble = true)
-    {
-        parent::__construct($level, $bubble);
-    }
-
    public function clearBuffer(): array
    {
        $logs = $this->buffer;
@@ -28,7 +23,7 @@ class BufferHandler extends AbstractProcessingHandler
        return $logs;
    }

-    protected function write(array $record): void
+    protected function write(array|LogRecord $record): void
    {
        $this->buffer[] = $record['formatted'];
    }
--- a/src/Logger/DelegatingLogger.php
+++ b/src/Logger/DelegatingLogger.php
@@ -16,7 +16,7 @@ final class DelegatingLogger extends AbstractLogger
    {
        foreach ($loggers as $logger) {
            if (!$logger instanceof LoggerInterface) {
-                throw new \InvalidArgumentException(sprintf('Only instances of %s should be passed to the constructor of %s. An instance of %s was passed instead.', LoggerInterface::class, self::class, $logger::class));
+                throw new \InvalidArgumentException(\sprintf('Only instances of %s should be passed to the constructor of %s. An instance of %s was passed instead.', LoggerInterface::class, self::class, $logger::class));
            }

            $this->loggers[] = $logger;
--- a/src/Manager/ScheduledDataflowManager.php
+++ b/src/Manager/ScheduledDataflowManager.php
@@ -19,15 +19,12 @@ class ScheduledDataflowManager implements ScheduledDataflowManagerInterface
    {
    }

-    /**
-     * {@inheritdoc}
-     */
    public function createJobsFromScheduledDataflows(): void
    {
        $this->connection->beginTransaction();
        try {
            foreach ($this->scheduledDataflowRepository->findReadyToRun() as $scheduled) {
-                if (null !== $this->jobRepository->findPendingForScheduledDataflow($scheduled)) {
+                if ($this->jobRepository->findPendingForScheduledDataflow($scheduled) !== null) {
                    continue;
                }

--- a/src/MessengerMode/JobMessageHandler.php
+++ b/src/MessengerMode/JobMessageHandler.php
@@ -6,9 +6,10 @@ namespace CodeRhapsodie\DataflowBundle\MessengerMode;

 use CodeRhapsodie\DataflowBundle\Processor\JobProcessorInterface;
 use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
-use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;
+use Symfony\Component\Messenger\Attribute\AsMessageHandler;

-class JobMessageHandler implements MessageSubscriberInterface
+#[AsMessageHandler]
+class JobMessageHandler
 {
    public function __construct(private JobRepository $repository, private JobProcessorInterface $processor)
    {
@@ -18,9 +19,4 @@ class JobMessageHandler implements MessageSubscriberInterface
    {
        $this->processor->process($this->repository->find($message->getJobId()));
    }
-
-    public static function getHandledMessages(): iterable
-    {
-        return [JobMessage::class];
-    }
 }
--- a/src/Processor/JobProcessor.php
+++ b/src/Processor/JobProcessor.php
@@ -4,6 +4,7 @@ declare(strict_types=1);

 namespace CodeRhapsodie\DataflowBundle\Processor;

+use CodeRhapsodie\DataflowBundle\DataflowType\RepositoryInterface;
 use CodeRhapsodie\DataflowBundle\DataflowType\Result;
 use CodeRhapsodie\DataflowBundle\Entity\Job;
 use CodeRhapsodie\DataflowBundle\Event\Events;
@@ -30,6 +31,10 @@ class JobProcessor implements JobProcessorInterface, LoggerAwareInterface
        $this->beforeProcessing($job);

        $dataflowType = $this->registry->getDataflowType($job->getDataflowType());
+        if ($dataflowType instanceof RepositoryInterface) {
+            $dataflowType->setRepository($this->repository);
+        }
+
        $loggers = [new Logger('dataflow_internal', [$bufferHandler = new BufferHandler()])];
        if (isset($this->logger)) {
            $loggers[] = $this->logger;
@@ -40,7 +45,7 @@ class JobProcessor implements JobProcessorInterface, LoggerAwareInterface
            $dataflowType->setLogger($logger);
        }

-        $result = $dataflowType->process($job->getOptions());
+        $result = $dataflowType->process($job->getOptions(), $job->getId());

        if (!$dataflowType instanceof LoggerAwareInterface) {
            foreach ($result->getExceptions() as $index => $e) {
@@ -53,12 +58,7 @@ class JobProcessor implements JobProcessorInterface, LoggerAwareInterface

    private function beforeProcessing(Job $job): void
    {
-        // Symfony 3.4 to 4.4 call
-        if (!class_exists(\Symfony\Contracts\EventDispatcher\Event::class)) {
-            $this->dispatcher->dispatch(Events::BEFORE_PROCESSING, new ProcessingEvent($job));
-        } else { // Symfony 5.0+ call
-            $this->dispatcher->dispatch(new ProcessingEvent($job), Events::BEFORE_PROCESSING);
-        }
+        $this->dispatcher->dispatch(new ProcessingEvent($job), Events::BEFORE_PROCESSING);

        $job
            ->setStatus(Job::STATUS_RUNNING)
@@ -77,11 +77,6 @@ class JobProcessor implements JobProcessorInterface, LoggerAwareInterface
        ;
        $this->repository->save($job);

-        // Symfony 3.4 to 4.4 call
-        if (!class_exists(\Symfony\Contracts\EventDispatcher\Event::class)) {
-            $this->dispatcher->dispatch(Events::AFTER_PROCESSING, new ProcessingEvent($job));
-        } else { // Symfony 5.0+ call
-            $this->dispatcher->dispatch(new ProcessingEvent($job), Events::AFTER_PROCESSING);
-        }
+        $this->dispatcher->dispatch(new ProcessingEvent($job), Events::AFTER_PROCESSING);
    }
 }
--- a/src/Registry/DataflowTypeRegistry.php
+++ b/src/Registry/DataflowTypeRegistry.php
@@ -18,9 +18,6 @@ class DataflowTypeRegistry implements DataflowTypeRegistryInterface
    /** @var array|DataflowTypeInterface[] */
    private array $aliasesRegistry = [];

-    /**
-     * {@inheritdoc}
-     */
    public function getDataflowType(string $fqcnOrAlias): DataflowTypeInterface
    {
        if (isset($this->fqcnRegistry[$fqcnOrAlias])) {
@@ -34,17 +31,11 @@ class DataflowTypeRegistry implements DataflowTypeRegistryInterface
        throw UnknownDataflowTypeException::create($fqcnOrAlias, [...array_keys($this->fqcnRegistry), ...array_keys($this->aliasesRegistry)]);
    }

-    /**
-     * {@inheritdoc}
-     */
    public function listDataflowTypes(): iterable
    {
        return $this->fqcnRegistry;
    }

-    /**
-     * {@inheritdoc}
-     */
    public function registerDataflowType(DataflowTypeInterface $dataflowType): void
    {
        $this->fqcnRegistry[$dataflowType::class] = $dataflowType;
--- a/src/Repository/InitFromDbTrait.php
+++ b/src/Repository/InitFromDbTrait.php
@@ -9,10 +9,12 @@ namespace CodeRhapsodie\DataflowBundle\Repository;
 */
 trait InitFromDbTrait
 {
+    abstract private function getFields(): array;
+
    private function initDateTime(array $datas): array
    {
-        foreach (static::FIELDS_TYPE as $key => $type) {
-            if ('datetime' === $type && null !== $datas[$key]) {
+        foreach ($this->getFields() as $key => $type) {
+            if ($type === 'datetime' && $datas[$key] !== null) {
                $datas[$key] = new \DateTime($datas[$key]);
            }
        }
@@ -22,10 +24,10 @@ trait InitFromDbTrait

    private function initArray(array $datas): array
    {
-        if (!is_array($datas['options'])) {
+        if (!\is_array($datas['options'])) {
            $datas['options'] = $this->strToArray($datas['options']);
        }
-        if (array_key_exists('exceptions', $datas) && !is_array($datas['exceptions'])) {
+        if (\array_key_exists('exceptions', $datas) && !\is_array($datas['exceptions'])) {
            $datas['exceptions'] = $this->strToArray($datas['exceptions']);
        }

@@ -34,12 +36,12 @@ trait InitFromDbTrait

    private function strToArray($value): array
    {
-        if (null === $value) {
+        if ($value === null) {
            return [];
        }

-        $array = json_decode($value, true, 512, JSON_THROW_ON_ERROR);
+        $array = json_decode($value, true, 512, \JSON_THROW_ON_ERROR);

-        return (false === $array) ? [] : $array;
+        return ($array === false) ? [] : $array;
    }
 }
--- a/src/Repository/JobRepository.php
+++ b/src/Repository/JobRepository.php
@@ -21,20 +21,6 @@ class JobRepository

    public const TABLE_NAME = 'cr_dataflow_job';

-    private const FIELDS_TYPE = [
-        'id' => ParameterType::INTEGER,
-        'status' => ParameterType::INTEGER,
-        'label' => ParameterType::STRING,
-        'dataflow_type' => ParameterType::STRING,
-        'options' => ParameterType::STRING,
-        'requested_date' => 'datetime',
-        'scheduled_dataflow_id' => ParameterType::INTEGER,
-        'count' => ParameterType::INTEGER,
-        'exceptions' => ParameterType::STRING,
-        'start_time' => 'datetime',
-        'end_time' => 'datetime',
-    ];
-
    /**
     * JobRepository constructor.
     */
@@ -58,8 +44,8 @@ class JobRepository
        $qb
            ->andWhere($qb->expr()->isNull('scheduled_dataflow_id'))
            ->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, ParameterType::INTEGER)));
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
+        $stmt = $qb->executeQuery();
+        if ($stmt->rowCount() === 0) {
            return [];
        }
        while (false !== ($row = $stmt->fetchAssociative())) {
@@ -106,8 +92,8 @@ class JobRepository
        $qb
            ->orderBy('requested_date', 'DESC')
            ->setMaxResults(20);
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
+        $stmt = $qb->executeQuery();
+        if ($stmt->rowCount() === 0) {
            return [];
        }
        while (false !== ($row = $stmt->fetchAssociative())) {
@@ -121,8 +107,8 @@ class JobRepository
        $qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($id, ParameterType::INTEGER)))
            ->orderBy('requested_date', 'DESC')
            ->setMaxResults(20);
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
+        $stmt = $qb->executeQuery();
+        if ($stmt->rowCount() === 0) {
            return [];
        }
        while (false !== ($row = $stmt->fetchAssociative())) {
@@ -135,20 +121,25 @@ class JobRepository
        $datas = $job->toArray();
        unset($datas['id']);

-        if (is_array($datas['options'])) {
-            $datas['options'] = json_encode($datas['options'], JSON_THROW_ON_ERROR);
+        if (\is_array($datas['options'])) {
+            $datas['options'] = json_encode($datas['options'], \JSON_THROW_ON_ERROR);
        }
-        if (is_array($datas['exceptions'])) {
-            $datas['exceptions'] = json_encode($datas['exceptions'], JSON_THROW_ON_ERROR);
+        if (\is_array($datas['exceptions'])) {
+            $datas['exceptions'] = json_encode($datas['exceptions'], \JSON_THROW_ON_ERROR);
        }

-        if (null === $job->getId()) {
-            $this->connection->insert(static::TABLE_NAME, $datas, static::FIELDS_TYPE);
+        if ($job->getId() === null) {
+            $this->connection->insert(static::TABLE_NAME, $datas, $this->getFields());
            $job->setId((int) $this->connection->lastInsertId());

            return;
        }
-        $this->connection->update(static::TABLE_NAME, $datas, ['id' => $job->getId()], static::FIELDS_TYPE);
+        $this->connection->update(static::TABLE_NAME, $datas, ['id' => $job->getId()], $this->getFields());
+    }
+
+    public function updateCount(int $jobId, int $count): void
+    {
+        $this->connection->update(static::TABLE_NAME, ['count' => $count], ['id' => $jobId]);
    }

    public function createQueryBuilder($alias = null): QueryBuilder
@@ -162,11 +153,28 @@ class JobRepository

    private function returnFirstOrNull(QueryBuilder $qb): ?Job
    {
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
+        $stmt = $qb->executeQuery();
+        if ($stmt->rowCount() === 0) {
            return null;
        }

        return Job::createFromArray($this->initDateTime($this->initArray($stmt->fetchAssociative())));
    }
+
+    private function getFields(): array
+    {
+        return [
+            'id' => ParameterType::INTEGER,
+            'status' => ParameterType::INTEGER,
+            'label' => ParameterType::STRING,
+            'dataflow_type' => ParameterType::STRING,
+            'options' => ParameterType::STRING,
+            'requested_date' => 'datetime',
+            'scheduled_dataflow_id' => ParameterType::INTEGER,
+            'count' => ParameterType::INTEGER,
+            'exceptions' => ParameterType::STRING,
+            'start_time' => 'datetime',
+            'end_time' => 'datetime',
+        ];
+    }
 }
--- a/src/Repository/ScheduledDataflowRepository.php
+++ b/src/Repository/ScheduledDataflowRepository.php
@@ -20,16 +20,6 @@ class ScheduledDataflowRepository

    public const TABLE_NAME = 'cr_dataflow_scheduled';

-    private const FIELDS_TYPE = [
-        'id' => ParameterType::INTEGER,
-        'label' => ParameterType::STRING,
-        'dataflow_type' => ParameterType::STRING,
-        'options' => ParameterType::STRING,
-        'frequency' => ParameterType::STRING,
-        'next' => 'datetime',
-        'enabled' => ParameterType::BOOLEAN,
-    ];
-
    /**
     * JobRepository constructor.
     */
@@ -46,12 +36,12 @@ class ScheduledDataflowRepository
    {
        $qb = $this->createQueryBuilder();
        $qb->andWhere($qb->expr()->lte('next', $qb->createNamedParameter(new \DateTime(), 'datetime')))
-            ->andWhere($qb->expr()->eq('enabled', 1))
+            ->andWhere($qb->expr()->eq('enabled', $qb->createNamedParameter(1, ParameterType::INTEGER)))
            ->orderBy('next', 'ASC')
        ;

-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
+        $stmt = $qb->executeQuery();
+        if ($stmt->rowCount() === 0) {
            return [];
        }
        while (false !== ($row = $stmt->fetchAssociative())) {
@@ -74,8 +64,8 @@ class ScheduledDataflowRepository
        $qb = $this->createQueryBuilder();
        $qb->orderBy('label', 'ASC');

-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
+        $stmt = $qb->executeQuery();
+        if ($stmt->rowCount() === 0) {
            return [];
        }
        while (false !== ($row = $stmt->fetchAssociative())) {
@@ -92,7 +82,7 @@ class ScheduledDataflowRepository
            ->orderBy('w.label', 'ASC')
            ->groupBy('w.id');

-        return $query->execute()->fetchAllAssociative();
+        return $query->executeQuery()->fetchAllAssociative();
    }

    public function save(ScheduledDataflow $scheduledDataflow)
@@ -100,17 +90,17 @@ class ScheduledDataflowRepository
        $datas = $scheduledDataflow->toArray();
        unset($datas['id']);

-        if (is_array($datas['options'])) {
-            $datas['options'] = json_encode($datas['options'], JSON_THROW_ON_ERROR);
+        if (\is_array($datas['options'])) {
+            $datas['options'] = json_encode($datas['options'], \JSON_THROW_ON_ERROR);
        }

-        if (null === $scheduledDataflow->getId()) {
-            $this->connection->insert(static::TABLE_NAME, $datas, static::FIELDS_TYPE);
+        if ($scheduledDataflow->getId() === null) {
+            $this->connection->insert(static::TABLE_NAME, $datas, $this->getFields());
            $scheduledDataflow->setId((int) $this->connection->lastInsertId());

            return;
        }
-        $this->connection->update(static::TABLE_NAME, $datas, ['id' => $scheduledDataflow->getId()], static::FIELDS_TYPE);
+        $this->connection->update(static::TABLE_NAME, $datas, ['id' => $scheduledDataflow->getId()], $this->getFields());
    }

    public function delete(int $id): void
@@ -138,11 +128,24 @@ class ScheduledDataflowRepository

    private function returnFirstOrNull(QueryBuilder $qb): ?ScheduledDataflow
    {
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
+        $stmt = $qb->executeQuery();
+        if ($stmt->rowCount() === 0) {
            return null;
        }

        return ScheduledDataflow::createFromArray($this->initDateTime($this->initArray($stmt->fetchAssociative())));
    }
+
+    private function getFields(): array
+    {
+        return [
+            'id' => ParameterType::INTEGER,
+            'label' => ParameterType::STRING,
+            'dataflow_type' => ParameterType::STRING,
+            'options' => ParameterType::STRING,
+            'frequency' => ParameterType::STRING,
+            'next' => 'datetime',
+            'enabled' => ParameterType::BOOLEAN,
+        ];
+    }
 }
--- a/src/Resources/config/services.yaml
+++ b/src/Resources/config/services.yaml
@@ -23,6 +23,7 @@ services:
        arguments:
            $registry: '@CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface'
            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
+            $jobRepository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
        tags: ['console.command']

    CodeRhapsodie\DataflowBundle\Command\JobShowCommand:
@@ -45,9 +46,15 @@ services:
        tags: ['console.command']

    CodeRhapsodie\DataflowBundle\Command\SchemaCommand:
+        deprecated:
+            package: 'code-rhapsodie/dataflow-bundle'
+            version: '5.0'
+        tags: ['console.command']
+
+    CodeRhapsodie\DataflowBundle\Command\DatabaseSchemaCommand:
        arguments:
            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
-        tags: ['console.command']
+        tags: [ 'console.command' ]

    CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository:
        lazy: true
--- a/src/Runner/PendingDataflowRunner.php
+++ b/src/Runner/PendingDataflowRunner.php
@@ -13,9 +13,6 @@ class PendingDataflowRunner implements PendingDataflowRunnerInterface
    {
    }

-    /**
-     * {@inheritdoc}
-     */
    public function runPendingDataflows(): void
    {
        while (null !== ($job = $this->repository->findNextPendingDataflow())) {
--- a/src/SchemaProvider/DataflowSchemaProvider.php
+++ b/src/SchemaProvider/DataflowSchemaProvider.php
@@ -47,7 +47,8 @@ class DataflowSchemaProvider
        $tableSchedule->addColumn('next', 'datetime', ['notnull' => false]);
        $tableSchedule->addColumn('enabled', 'boolean', ['notnull' => true]);

-        $tableJob->addForeignKeyConstraint($tableSchedule, ['scheduled_dataflow_id'], ['id']);
+        $tableJob->addForeignKeyConstraint($tableSchedule->getName(), ['scheduled_dataflow_id'], ['id']);
+        $tableJob->addIndex(['status'], 'idx_status');

        return $schema;
    }
--- a/src/Validator/Constraints/FrequencyValidator.php
+++ b/src/Validator/Constraints/FrequencyValidator.php
@@ -10,20 +10,22 @@ use Symfony\Component\Validator\Exception\UnexpectedTypeException;

 class FrequencyValidator extends ConstraintValidator
 {
-    /**
-     * {@inheritdoc}
-     */
-    public function validate($value, Constraint $constraint)
+    public function validate(mixed $value, Constraint $constraint)
    {
        if (!$constraint instanceof Frequency) {
            throw new UnexpectedTypeException($constraint, Frequency::class);
        }

-        if (null === $value) {
+        if ($value === null) {
            return;
        }

-        $interval = @\DateInterval::createFromDateString($value);
+        try {
+            $interval = \DateInterval::createFromDateString($value);
+        } catch (\Exception) {
+            $interval = false;
+        }
+
        if (!$interval) {
            $this->context->buildViolation($constraint->message)
                ->setParameter('{{ string }}', $value)
@@ -42,8 +44,6 @@ class FrequencyValidator extends ConstraintValidator
                ->setParameter('{{ string }}', $value)
                ->addViolation()
            ;
-
-            return;
        }
    }
 }
Author	SHA1	Message	Date
AUDUL	4e82b114c4	Add auto update count processed item while running job (#79 ) * Add auto update count processed item while running job	2025-10-02 15:22:37 +02:00
AUDUL	47af0e226c	Added custom index for job status (#77 ) * Added custom index for job status	2025-07-04 09:26:51 +02:00
AUDUL	6d86ba16a0	SchemaDump command (#75 ) * Added possibility to install/update database from command #37	2024-10-31 16:40:09 +01:00
AUDUL	95124acc26	* Fix compatibility with doctrine 4 (#73 ) * Remove some deprecated	2024-10-31 15:38:10 +01:00
AUDUL	4011f39510	V5 (#72 ) * Added Symfony 7 support * Removed Symfony 6 compatibility * Removed Symfony 5 compatibility * Removed Symfony 4 compatibility * Removed Symfony 3 compatibility * Changed README.md * Added CI --------- Co-authored-by: Jérémy J <jeremy@code-rhapsodie.fr>	2024-10-31 13:01:43 +01:00
Matt Mankins	db37c4bdd1	Update Kudos Github Action to support generation from source repo only (#71 ) * Update semicolons-kudos.yaml * Update GitHub Action workflow for Semicolons Kudos Action * Update semicolons-kudos.yaml * Update GitHub Action workflow for Semicolons Kudos Action * Update GitHub Action workflow for Semicolons Kudos Action --------- Co-authored-by: semicolons-for-kudos[bot] <145267638+semicolons-for-kudos[bot]@users.noreply.github.com>	2023-12-27 17:29:46 +01:00
Olivier PORTIER	f20cd96ec5	Update semicolons-kudos.yaml	2023-12-20 11:49:50 +01:00
Olivier PORTIER	fd2c6aaab5	Update semicolons-kudos.yaml (#70 )	2023-12-20 11:36:51 +01:00
Olivier PORTIER	4efd310a6e	Initiate Kudos on dataflow-bundle by creating new file semicolons-kudos.yaml (#69 )	2023-12-20 11:11:26 +01:00
Jérémy J	cec42a3337	Fix log exception argument typing	2023-12-06 13:56:16 +01:00
jbcr	d440ad008b	add sonar config	2023-11-16 16:50:24 +01:00