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>

.coveralls.yml

@@ -1,3 +0,0 @@
-service_name   : travis-ci
-coverage_clover: var/build/clover.xml
-json_path      : var/build/upload.json

.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 }}

.github/workflows/ci.yml

@@ -0,0 +1,12 @@
+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

.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: ""

.gitignore

@@ -3,3 +3,6 @@ composer.lock
 .phpunit.result.cache
 .php_cs.cache
 .php_cs
+.idea
+.phpunit.cache
+.php-version

.travis.yml

@@ -1,135 +0,0 @@
-language: php
-
-sudo: false
-
-cache:
-  directories:
-    - $HOME/.composer/cache
-
-branches:
-  only:
-    - master
-    - /^\d+\.\d+$/
-    - travis-setup
-
-env:
-  global:
-    - SYMFONY_DEPRECATIONS_HELPER="max[self]=0"
-    - PHPUNIT_FLAGS="-v"
-    - PHPUNIT_ENABLED="true"
-    - STABILITY=stable
-    - COVERALLS_ENABLED="false"
-
-matrix:
-  fast_finish: true
-  include:
-    - php: '7.3'
-    - php: '7.4'
-    - php: '8.0'
-
-    # Enable code coverage with the previous supported PHP version
-#    - php: '7.4'
-#      env:
-#        - SYMFONY_VERSION=3.4.*
-#        - COVERALLS_ENABLED="true"
-#        - PHPUNIT_FLAGS="-v --coverage-text --coverage-clover var/build/clover.xml"
-
-    # Enable code coverage with the latest supported PHP version
-#    - php: '8.0'
-#      env:
-#        - SYMFONY_VERSION=3.4.*
-#        - COVERALLS_ENABLED="true"
-#        - PHPUNIT_FLAGS="-v --coverage-text --coverage-clover var/build/clover.xml"
-
-    # Minimum supported dependencies with the latest and oldest supported PHP versions
-    - php: '7.3'
-      env:
-        - COMPOSER_FLAGS="--prefer-lowest"
-    # Incompatibility between lowest symfony testing utils and phpunit
-#    - php: '8.0'
-#      env:
-#        - COMPOSER_FLAGS="--prefer-lowest"
-
-    # Test each supported Symfony version with lowest supported PHP version
-#    - php: '7.3'
-#      env:
-#        - SYMFONY_VERSION=3.4.*
-    - php: '7.3'
-      env:
-        - SYMFONY_VERSION=4.4.*
-    - php: '7.3'
-      env:
-        - COVERALLS_ENABLED="true"
-        - PHPUNIT_FLAGS="-v --coverage-text --coverage-clover var/build/clover.xml"
-        - SYMFONY_VERSION=5.2.*
-    # Test unsupported versions of Symfony
-#    - php: '7.3'
-#      env:
-#        - SYMFONY_VERSION=4.1.*
-#    - php: '7.3'
-#      env:
-#        - SYMFONY_VERSION=4.2.*
-#    - php: '7.3'
-#      env:
-#        - SYMFONY_VERSION=4.3.*
-    - php: '7.3'
-      env:
-        - SYMFONY_VERSION=5.0.*
-    - php: '7.3'
-      env:
-        - SYMFONY_VERSION=5.1.*
-
-    # Test upcoming Symfony versions with lowest supported PHP version and dev dependencies
-#    - php: '7.2'
-#      env:
-#        - STABILITY=dev
-#        - SYMFONY_VERSION=5.3.*
-
-    # Test upcoming PHP versions with dev dependencies
-    #- php: '7.5snapshot'
-    #  env:
-    #    - STABILITY=dev
-    #    - COMPOSER_FLAGS="--ignore-platform-reqs --prefer-stable"
-
-  allow_failures:
-    # 4.0 not supported because of https://github.com/advisories/GHSA-pgwj-prpq-jpc2
-    - env:
-        - SYMFONY_VERSION=4.0.*
-    - env:
-        - SYMFONY_VERSION=4.1.*
-    - env:
-        - SYMFONY_VERSION=4.2.*
-    - env:
-        - STABILITY=dev
-        - COMPOSER_FLAGS="--ignore-platform-reqs --prefer-stable"
-    - env:
-        - STABILITY=dev
-        - SYMFONY_VERSION=5.1.*
-    - env:
-        - STABILITY=dev
-        - SYMFONY_VERSION=5.2.*
-
-before_install:
-  - if [[ "$SYMFONY_VERSION" != "" ]]; then
-      travis_retry composer global require "symfony/flex:^1.4";
-      composer config extra.symfony.require $SYMFONY_VERSION;
-    fi
-  - if [[ "$STABILITY" != "stable" ]]; then
-      travis_retry composer config minimum-stability $STABILITY;
-    fi
-  - if [[ "$COVERALLS_ENABLED" != "true" ]]; then
-      phpenv config-rm xdebug.ini || true;
-    fi
-  - if [[ "$COVERALLS_ENABLED" == "true" ]]; then
-      travis_retry composer require --dev php-coveralls/php-coveralls:^2.0 --no-update $COMPOSER_FLAGS;
-    fi
-
-install:
-  - travis_retry composer update --prefer-dist --no-interaction --no-progress --ansi $COMPOSER_FLAGS
-
-script: ./vendor/bin/phpunit $PHPUNIT_FLAGS
-
-after_success:
-  - if [[ "$PHPUNIT_ENABLED" == "true" && "$COVERALLS_ENABLED" == "true" ]]; then
-    ./vendor/bin/php-coveralls -vvv --config .coveralls.yml;
-    fi;

CHANGELOG.md

@@ -1,3 +1,12 @@
+# Version 4.1.0
+
+* Added custom index for exception log
+
+# Version 4.0.0
+
+* Added Symfony 6 support
+* PHP minimum requirements bumped to 8.0
+
 # Version 3.1.0
 
 * Added optional "messenger mode", to delegate jobs execution to workers from the Symfony messenger component

README.md

@@ -1,18 +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.
 
-[![Build Status](https://travis-ci.com/code-rhapsodie/dataflow-bundle.svg?branch=master)](https://travis-ci.com/code-rhapsodie/dataflow-bundle)
-
-[![Coverage Status](https://coveralls.io/repos/github/code-rhapsodie/dataflow-bundle/badge.svg)](https://coveralls.io/github/code-rhapsodie/dataflow-bundle)
+| 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      |
 
 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. 
+* 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.
 
@@ -20,7 +25,6 @@ As the following schema shows, you can define more than one dataflow:
 
 ![Dataflow schema](src/Resources/doc/schema.png)
 
-
 # Features
 
 * Define and configure a Dataflow
@@ -32,13 +36,10 @@ 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
 
-And basically, every allowed-to-failed jobs in our travis configuration are not fully supported.
-
 ### Add the dependency
 
 To install this bundle, run this command :
@@ -51,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
@@ -64,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:
@@ -81,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:
 
@@ -144,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:
@@ -152,6 +134,7 @@ code_rhapsodie_dataflow:
 ```
 
 You also need to route Dataflow messages to the proper transport:
+
 ```yaml
 # config/packages/messenger.yaml
 framework:
@@ -165,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*
@@ -176,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:
 
@@ -237,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
@@ -271,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
 
@@ -299,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:
@@ -323,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`.
 
@@ -364,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
 
@@ -416,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 :
 
@@ -467,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));
@@ -477,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
@@ -552,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
 
@@ -570,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.
 
@@ -609,6 +609,8 @@ $ 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 4.x and the 5.x versions of this bundle are actively supported.
+
 # Contributing
 
 Contributions are very welcome. Please see [CONTRIBUTING.md](CONTRIBUTING.md) for

Tests/DataflowType/AbstractDataflowTypeTest.php

@@ -18,17 +18,8 @@ class AbstractDataflowTypeTest extends TestCase
 
         $dataflowType = new class($label, $options, $values, $testCase) extends AbstractDataflowType
         {
-            private $label;
-            private $options;
-            private $values;
-            private $testCase;
-
-            public function __construct(string $label, array $options, array $values, TestCase $testCase)
+            public function __construct(private string $label, private array $options, private array $values, private TestCase $testCase)
             {
-                $this->label = $label;
-                $this->options = $options;
-                $this->values = $values;
-                $this->testCase = $testCase;
             }
 
             public function getLabel(): string

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;
 
@@ -15,9 +14,7 @@ class AMPAsyncDataflowTest extends TestCase
         $reader = [1, 2, 3];
         $result = [];
         $dataflow = new AMPAsyncDataflow($reader, 'simple');
-        $dataflow->addStep(static function($item) {
-            return $item + 1;
-        });
+        $dataflow->addStep(static fn($item) => $item + 1);
         $dataflow->addStep(static function($item): \Generator {
             yield new Delayed(10); //delay 10 milliseconds
             return $item * 2;

Tests/DataflowType/Writer/CollectionWriterTest.php

@@ -40,12 +40,13 @@ 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(function ($item) {
-                return [$item];
-            }, $values))
+            ->with($this->callback(function ($arg) use ($matcher, $values) {
+                return $arg === $values[$matcher->numberOfInvocations() - 1];
+            }))
         ;
 
         $writer = new CollectionWriter($embeddedWriter);

Tests/DataflowType/Writer/DelegatorWriterTest.php

@@ -10,34 +10,21 @@ use PHPUnit\Framework\TestCase;
 
 class DelegatorWriterTest extends TestCase
 {
-    /** @var DelegatorWriter */
-    private $delegatorWriter;
-
-    /** @var DelegateWriterInterface|MockObject */
-    private $delegateInt;
-
-    /** @var DelegateWriterInterface|MockObject */
-    private $delegateString;
-
-    /** @var DelegateWriterInterface|MockObject */
-    private $delegateArray;
+    private DelegatorWriter $delegatorWriter;
+    private DelegateWriterInterface|MockObject $delegateInt;
+    private DelegateWriterInterface|MockObject $delegateString;
+    private DelegateWriterInterface|MockObject $delegateArray;
 
     protected function setUp(): void
     {
         $this->delegateInt = $this->createMock(DelegateWriterInterface::class);
-        $this->delegateInt->method('supports')->willReturnCallback(function ($argument) {
-            return is_int($argument);
-        });
+        $this->delegateInt->method('supports')->willReturnCallback(fn($argument) => is_int($argument));
 
         $this->delegateString = $this->createMock(DelegateWriterInterface::class);
-        $this->delegateString->method('supports')->willReturnCallback(function ($argument) {
-            return is_string($argument);
-        });
+        $this->delegateString->method('supports')->willReturnCallback(fn($argument) => is_string($argument));
 
         $this->delegateArray = $this->createMock(DelegateWriterInterface::class);
-        $this->delegateArray->method('supports')->willReturnCallback(function ($argument) {
-            return is_array($argument);
-        });
+        $this->delegateArray->method('supports')->willReturnCallback(fn($argument) => is_array($argument));
 
         $this->delegatorWriter = new DelegatorWriter();
         $this->delegatorWriter->addDelegates([

Tests/DataflowType/Writer/PortWriterAdapterTest.php

@@ -12,7 +12,7 @@ class PortWriterAdapterTest extends TestCase
         $value = 'not an array';
 
         $writer = $this->getMockBuilder('\Port\Writer')
-            ->setMethods(['prepare', 'finish', 'writeItem'])
+            ->onlyMethods(['prepare', 'finish', 'writeItem'])
             ->getMock()
         ;
         $writer

Tests/Manager/ScheduledDataflowManagerTest.php

@@ -2,31 +2,21 @@
 
 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;
-use Doctrine\DBAL\Driver\Connection;
-use Doctrine\ORM\EntityManagerInterface;
+use Doctrine\DBAL\Connection;
 use PHPUnit\Framework\MockObject\MockObject;
 use PHPUnit\Framework\TestCase;
 
 class ScheduledDataflowManagerTest extends TestCase
 {
-    /** @var ScheduledDataflowManager */
-    private $manager;
-
-    /** @var Connection|MockObject */
-    private $connection;
-
-    /** @var ScheduledDataflowRepository|MockObject */
-    private $scheduledDataflowRepository;
-
-    /** @var JobRepository|MockObject */
-    private $jobRepository;
+    private ScheduledDataflowManager $manager;
+    private Connection|MockObject $connection;
+    private ScheduledDataflowRepository|MockObject $scheduledDataflowRepository;
+    private JobRepository|MockObject $jobRepository;
 
     protected function setUp(): void
     {
@@ -55,10 +45,20 @@ 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(function ($arg) use ($matcher, $scheduled1, $scheduled2) {
+                switch ($matcher->numberOfInvocations()) {
+                    case 1:
+                        return $arg === $scheduled1;
+                    case 2:
+                        return $arg === $scheduled2;
+                    default:
+                        return false;
+                }
+            }))
             ->willReturnOnConsecutiveCalls(new Job(), null)
         ;
 
@@ -70,16 +70,12 @@ class ScheduledDataflowManagerTest extends TestCase
             ->expects($this->once())
             ->method('save')
             ->with(
-                $this->callback(function (Job $job) use ($type, $options, $next, $label, $scheduled2) {
-                    return (
-                        $job->getStatus() === Job::STATUS_PENDING
-                        && $job->getDataflowType() === $type
-                        && $job->getOptions() === $options
-                        && $job->getRequestedDate() == $next
-                        && $job->getLabel() === $label
-                        && $job->getScheduledDataflowId() === $scheduled2->getId()
-                    );
-                })
+                $this->callback(fn(Job $job) => $job->getStatus() === Job::STATUS_PENDING
+                && $job->getDataflowType() === $type
+                && $job->getOptions() === $options
+                && $job->getRequestedDate() == $next
+                && $job->getLabel() === $label
+                && $job->getScheduledDataflowId() === $scheduled2->getId())
             )
         ;
 
@@ -112,7 +108,7 @@ class ScheduledDataflowManagerTest extends TestCase
         $this->jobRepository
             ->expects($this->exactly(1))
             ->method('findPendingForScheduledDataflow')
-            ->withConsecutive([$scheduled1])
+            ->with($scheduled1)
             ->willThrowException(new \Exception())
         ;
 

Tests/MessengerMode/JobMessageHandlerTest.php

@@ -14,14 +14,9 @@ use PHPUnit\Framework\TestCase;
 
 class JobMessageHandlerTest extends TestCase
 {
-    /** @var JobRepository|MockObject */
-    private $repository;
-
-    /** @var JobProcessorInterface|MockObject */
-    private $processor;
-
-    /** @var JobMessageHandler */
-    private $handler;
+    private JobRepository|MockObject $repository;
+    private JobProcessorInterface|MockObject $processor;
+    private JobMessageHandler $handler;
 
     protected function setUp(): void
     {
@@ -31,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);

Tests/Processor/JobProcessorTest.php

@@ -16,17 +16,10 @@ use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
 class JobProcessorTest extends TestCase
 {
-    /** @var JobProcessor */
-    private $processor;
-
-    /** @var JobRepository|MockObject */
-    private $repository;
-
-    /** @var DataflowTypeRegistryInterface|MockObject */
-    private $registry;
-
-    /** @var EventDispatcherInterface|MockObject */
-    private $dispatcher;
+    private JobProcessor $processor;
+    private JobRepository|MockObject $repository;
+    private DataflowTypeRegistryInterface|MockObject $registry;
+    private EventDispatcherInterface|MockObject $dispatcher;
 
     protected function setUp(): void
     {
@@ -46,44 +39,25 @@ class JobProcessorTest extends TestCase
             ->setOptions($options = ['option1' => 'value1'])
         ;
 
-        // Symfony 3.4 to 4.4 call
-        if (!class_exists('Symfony\Contracts\EventDispatcher\Event')) {
-            $this->dispatcher
-                ->expects($this->exactly(2))
-                ->method('dispatch')
-                ->withConsecutive(
-                    [
-                        Events::BEFORE_PROCESSING,
-                        $this->callback(function (ProcessingEvent $event) use ($job) {
-                            return $event->getJob() === $job;
-                        })
-                    ],
-                    [
-                        Events::AFTER_PROCESSING,
-                        $this->callback(function (ProcessingEvent $event) use ($job) {
-                            return $event->getJob() === $job;
-                        })
-                    ],
-                );
-        } else { // Symfony 5.0+
-            $this->dispatcher
-                ->expects($this->exactly(2))
-                ->method('dispatch')
-                ->withConsecutive(
-                    [
-                        $this->callback(function (ProcessingEvent $event) use ($job) {
-                            return $event->getJob() === $job;
-                        }),
-                        Events::BEFORE_PROCESSING,
-                    ],
-                    [
-                        $this->callback(function (ProcessingEvent $event) use ($job) {
-                            return $event->getJob() === $job;
-                        }),
-                        Events::AFTER_PROCESSING,
-                    ],
-                );
-        }
+        $matcher = $this->exactly(2);
+        $this->dispatcher
+            ->expects($matcher)
+            ->method('dispatch')
+            ->with(
+                $this->callback(function ($arg) use ($job) {
+                    return $arg instanceof ProcessingEvent && $arg->getJob() === $job;
+                }),
+                $this->callback(function ($arg) use ($matcher) {
+                    switch ($matcher->numberOfInvocations()) {
+                        case 1:
+                            return $arg === Events::BEFORE_PROCESSING;
+                        case 2:
+                            return $arg === Events::AFTER_PROCESSING;
+                        default:
+                            return false;
+                    }
+                })
+            );
 
         $dataflowType = $this->createMock(DataflowTypeInterface::class);
 

Tests/Registry/DataflowTypeRegistryTest.php

@@ -10,8 +10,7 @@ use PHPUnit\Framework\TestCase;
 
 class DataflowTypeRegistryTest extends TestCase
 {
-    /** @var DataflowTypeRegistry */
-    private $registry;
+    private DataflowTypeRegistry $registry;
 
     protected function setUp(): void
     {
@@ -33,7 +32,7 @@ class DataflowTypeRegistryTest extends TestCase
 
         $this->registry->registerDataflowType($type);
 
-        $this->assertSame($type, $this->registry->getDataflowType(get_class($type)));
+        $this->assertSame($type, $this->registry->getDataflowType($type::class));
         $this->assertSame($type, $this->registry->getDataflowType($alias1));
         $this->assertSame($type, $this->registry->getDataflowType($alias2));
         $this->assertContains($type, $this->registry->listDataflowTypes());

Tests/Runner/MessengerDataflowRunnerTest.php

@@ -13,14 +13,9 @@ use Symfony\Component\Messenger\MessageBusInterface;
 
 class MessengerDataflowRunnerTest extends TestCase
 {
-    /** @var MessengerDataflowRunner */
-    private $runner;
-
-    /** @var JobRepository|MockObject */
-    private $repository;
-
-    /** @var MessageBusInterface|MockObject */
-    private $bus;
+    private MessengerDataflowRunner $runner;
+    private JobRepository|MockObject $repository;
+    private MessageBusInterface|MockObject $bus;
 
     protected function setUp(): void
     {
@@ -40,24 +35,36 @@ 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(function ($arg) use ($matcher, $job1, $job2) {
+                switch ($matcher->numberOfInvocations()) {
+                    case 1:
+                        return $arg === $job1;
+                    case 2:
+                        return $arg === $job2;
+                    default:
+                        return false;
+                }
+            }))
         ;
 
+        $matcher = $this->exactly(2);
         $this->bus
-            ->expects($this->exactly(2))
+            ->expects($matcher)
             ->method('dispatch')
-            ->withConsecutive([
-                $this->callback(function ($message) use ($id1) {
-                    return $message instanceof JobMessage && $message->getJobId() === $id1;
-                })
-            ], [
-                $this->callback(function ($message) use ($id2) {
-                    return $message instanceof JobMessage && $message->getJobId() === $id2;
-                })
-            ])
+            ->with($this->callback(function ($arg) use ($matcher, $id1, $id2) {
+                switch ($matcher->numberOfInvocations()) {
+                    case 1:
+                        return $arg instanceof JobMessage && $arg->getJobId() === $id1;
+                    case 2:
+                        return $arg instanceof JobMessage && $arg->getJobId() === $id2;
+                    default:
+                        return false;
+                }
+            }))
             ->willReturnOnConsecutiveCalls(
                 new Envelope(new JobMessage($id1)),
                 new Envelope(new JobMessage($id2))

Tests/Runner/PendingDataflowRunnerTest.php

@@ -11,14 +11,9 @@ use PHPUnit\Framework\TestCase;
 
 class PendingDataflowRunnerTest extends TestCase
 {
-    /** @var PendingDataflowRunner */
-    private $runner;
-
-    /** @var JobRepository|MockObject */
-    private $repository;
-
-    /** @var JobProcessorInterface|MockObject */
-    private $processor;
+    private PendingDataflowRunner $runner;
+    private JobRepository|MockObject $repository;
+    private JobProcessorInterface|MockObject $processor;
 
     protected function setUp(): void
     {
@@ -39,10 +34,20 @@ 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(function ($arg) use ($matcher, $job1, $job2) {
+                switch ($matcher->numberOfInvocations()) {
+                    case 1:
+                        return $arg === $job1;
+                    case 2:
+                        return $arg === $job2;
+                    default:
+                        return false;
+                }
+            }))
         ;
 
         $this->runner->runPendingDataflows();

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'],

composer.json

@@ -41,26 +41,29 @@
         }
     },
     "require": {
-        "php": "^7.3||^8.0",
+        "php": "^8.0",
         "ext-json": "*",
-        "doctrine/dbal": "^2.12||^3.0",
-        "doctrine/doctrine-bundle": "^1.0||^2.0",
-        "psr/log": "^1.1",
-        "symfony/config": "^3.4||^4.0||^5.0",
-        "symfony/console": "^3.4||^4.0||^5.0",
-        "symfony/dependency-injection": "^3.4||>=4.1.12||^5.0",
-        "symfony/event-dispatcher": "^3.4||^4.0||^5.0",
-        "symfony/http-kernel": "^3.4||^4.0||^5.0",
-        "symfony/lock": "^3.4||^4.0||^5.0",
-        "symfony/monolog-bridge": "^3.4||^4.0||^5.0",
-        "symfony/options-resolver": "^3.4||^4.0||^5.0",
-        "symfony/validator": "^3.4||^4.0||^5.0",
-        "symfony/yaml": "^3.4||^4.0||^5.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": "^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",
-        "symfony/messenger": "^4.4||^5.0"
+        "phpunit/phpunit": "^11",
+        "portphp/portphp": "^1.9",
+        "rector/rector": "^1.0",
+        "symfony/messenger": "^7.0"
     },
     "suggest": {
         "amphp/amp": "Provide asynchronous steps for your dataflows",

phpunit.xml

@@ -1,32 +1,21 @@
-<?xml version = '1.0' encoding = 'UTF-8'?>
-
+<?xml version="1.0" encoding="UTF-8"?>
 <!-- http://www.phpunit.de/manual/current/en/appendixes.configuration.html -->
-<phpunit
-        backupGlobals="false"
-        backupStaticAttributes="false"
-        bootstrap="Tests/bootstrap.php"
-        convertErrorsToExceptions="true"
-        convertNoticesToExceptions="true"
-        convertWarningsToExceptions="true"
-        colors="false"
->
-    <php>
-        <ini name="error_reporting" value="-1" />
-    </php>
-    <testsuites>
-        <testsuite name="Dataflow tests suite">
-            <directory suffix="Test.php">./Tests</directory>
-        </testsuite>
-    </testsuites>
-
-    <filter>
-        <whitelist>
-            <directory>./src/</directory>
-            <exclude>
-                <directory>Tests/</directory>
-                <directory>vendor/</directory>
-            </exclude>
-        </whitelist>
-    </filter>
-
+<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>
+  <testsuites>
+    <testsuite name="Dataflow tests suite">
+      <directory suffix="Test.php">./Tests</directory>
+    </testsuite>
+  </testsuites>
+  <source>
+    <include>
+      <directory>./src/</directory>
+    </include>
+    <exclude>
+      <directory>Tests/</directory>
+      <directory>vendor/</directory>
+    </exclude>
+  </source>
 </phpunit>

rector.php

@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+use Rector\CodeQuality\Rector\Class_\InlineConstructorDefaultToPropertyRector;
+use Rector\Config\RectorConfig;
+use Rector\Set\ValueObject\LevelSetList;
+use Rector\Symfony\Set\SymfonySetList;
+
+return static function (RectorConfig $rectorConfig): void {
+    $rectorConfig->paths([
+        __DIR__ . '/src',
+        __DIR__ . '/Tests',
+    ]);
+
+    // register a single rule
+    $rectorConfig->rule(InlineConstructorDefaultToPropertyRector::class);
+
+    $rectorConfig->sets([
+        SymfonySetList::SYMFONY_60,
+        SymfonySetList::SYMFONY_CODE_QUALITY,
+        SymfonySetList::SYMFONY_CONSTRUCTOR_INJECTION,
+        LevelSetList::UP_TO_PHP_80,
+    ]);
+};

sonar-project.properties

@@ -0,0 +1,4 @@
+sonar.projectKey=code-rhapsodie_dataflow-bundle_AYvYuaAwWE9sbcQmD1vw
+
+sonar.sources=src
+sonar.tests=Tests

src/CodeRhapsodieDataflowBundle.php

@@ -9,6 +9,7 @@ use CodeRhapsodie\DataflowBundle\DependencyInjection\Compiler\BusCompilerPass;
 use CodeRhapsodie\DataflowBundle\DependencyInjection\Compiler\DataflowTypeCompilerPass;
 use CodeRhapsodie\DataflowBundle\DependencyInjection\Compiler\DefaultLoggerCompilerPass;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\DependencyInjection\Extension\ExtensionInterface;
 use Symfony\Component\HttpKernel\Bundle\Bundle;
 
 /**
@@ -16,9 +17,9 @@ use Symfony\Component\HttpKernel\Bundle\Bundle;
  */
 class CodeRhapsodieDataflowBundle extends Bundle
 {
-    protected $name = 'CodeRhapsodieDataflowBundle';
+    protected string $name = 'CodeRhapsodieDataflowBundle';
 
-    public function getContainerExtension()
+    public function getContainerExtension(): ?ExtensionInterface
     {
         return new CodeRhapsodieDataflowExtension();
     }

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,28 +19,12 @@ 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';
-
-    /** @var DataflowTypeRegistryInterface */
-    private $registry;
-    /** @var ScheduledDataflowRepository */
-    private $scheduledDataflowRepository;
-    /** @var ValidatorInterface */
-    private $validator;
-
-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(DataflowTypeRegistryInterface $registry, ScheduledDataflowRepository $scheduledDataflowRepository, ValidatorInterface $validator, ConnectionFactory $connectionFactory)
+    public function __construct(private DataflowTypeRegistryInterface $registry, private ScheduledDataflowRepository $scheduledDataflowRepository, private ValidatorInterface $validator, private ConnectionFactory $connectionFactory)
     {
         parent::__construct();
-
-        $this->registry = $registry;
-        $this->scheduledDataflowRepository = $scheduledDataflowRepository;
-        $this->validator = $validator;
-        $this->connectionFactory = $connectionFactory;
     }
 
     /**
@@ -48,7 +33,6 @@ class AddScheduledDataflowCommand extends Command
     protected function configure()
     {
         $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)')
@@ -63,7 +47,7 @@ class AddScheduledDataflowCommand extends Command
     /**
      * {@inheritdoc}
      */
-    protected function execute(InputInterface $input, OutputInterface $output)
+    protected function execute(InputInterface $input, OutputInterface $output): int
     {
         if (null !== $input->getOption('connection')) {
             $this->connectionFactory->setConnectionName($input->getOption('connection'));
@@ -108,9 +92,9 @@ class AddScheduledDataflowCommand extends Command
             'id' => null,
             'label' => $label,
             'dataflow_type' => $type,
-            'options' => json_decode($options, true),
+            'options' => json_decode($options, true, 512, JSON_THROW_ON_ERROR),
             'frequency' => $frequency,
-            'next' => new \DateTimeImmutable($firstRun),
+            'next' => new \DateTime($firstRun),
             'enabled' => $enabled,
         ]);
 

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,12 @@ 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';
-
-    /** @var ScheduledDataflowRepository */
-    private $scheduledDataflowRepository;
-
-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(ScheduledDataflowRepository $scheduledDataflowRepository, ConnectionFactory $connectionFactory)
+    public function __construct(private ScheduledDataflowRepository $scheduledDataflowRepository, private ConnectionFactory $connectionFactory)
     {
         parent::__construct();
-
-        $this->scheduledDataflowRepository = $scheduledDataflowRepository;
-        $this->connectionFactory = $connectionFactory;
     }
 
     /**
@@ -41,7 +32,6 @@ class ChangeScheduleStatusCommand extends Command
     protected function configure()
     {
         $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')
@@ -52,7 +42,7 @@ class ChangeScheduleStatusCommand extends Command
     /**
      * {@inheritdoc}
      */
-    protected function execute(InputInterface $input, OutputInterface $output)
+    protected function execute(InputInterface $input, OutputInterface $output): int
     {
         if (null !== $input->getOption('connection')) {
             $this->connectionFactory->setConnectionName($input->getOption('connection'));

src/Command/ExecuteDataflowCommand.php

@@ -8,6 +8,7 @@ use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
 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,24 +21,14 @@ 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';
-
-    /** @var DataflowTypeRegistryInterface */
-    private $registry;
-
-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(DataflowTypeRegistryInterface $registry, ConnectionFactory $connectionFactory)
+    public function __construct(private DataflowTypeRegistryInterface $registry, private ConnectionFactory $connectionFactory)
     {
         parent::__construct();
-
-        $this->registry = $registry;
-        $this->connectionFactory = $connectionFactory;
     }
 
     /**
@@ -46,7 +37,6 @@ class ExecuteDataflowCommand extends Command implements LoggerAwareInterface
     protected function configure()
     {
         $this
-            ->setDescription('Runs one dataflow type with provided options')
             ->setHelp(<<<'EOF'
 The <info>%command.name%</info> command runs one dataflow with the provided options.
 
@@ -61,13 +51,13 @@ EOF
     /**
      * {@inheritdoc}
      */
-    protected function execute(InputInterface $input, OutputInterface $output)
+    protected function execute(InputInterface $input, OutputInterface $output): int
     {
         if (null !== $input->getOption('connection')) {
             $this->connectionFactory->setConnectionName($input->getOption('connection'));
         }
         $fqcnOrAlias = $input->getArgument('fqcn');
-        $options = json_decode($input->getArgument('options'), true);
+        $options = json_decode($input->getArgument('options'), true, 512, JSON_THROW_ON_ERROR);
         $io = new SymfonyStyle($input, $output);
 
         $dataflowType = $this->registry->getDataflowType($fqcnOrAlias);

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,9 @@ class JobShowCommand extends Command
         Job::STATUS_COMPLETED => 'Completed',
     ];
 
-    protected static $defaultName = 'code-rhapsodie:dataflow:job:show';
-
-    /** @var JobRepository */
-    private $jobRepository;
-
-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(JobRepository $jobRepository, ConnectionFactory $connectionFactory)
+    public function __construct(private JobRepository $jobRepository, private ConnectionFactory $connectionFactory)
     {
         parent::__construct();
-
-        $this->jobRepository = $jobRepository;
-        $this->connectionFactory = $connectionFactory;
     }
 
     /**
@@ -46,7 +37,6 @@ class JobShowCommand extends Command
     protected function configure()
     {
         $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')
@@ -57,7 +47,7 @@ class JobShowCommand extends Command
     /**
      * {@inheritdoc}
      */
-    protected function execute(InputInterface $input, OutputInterface $output)
+    protected function execute(InputInterface $input, OutputInterface $output): int
     {
         if (null !== $input->getOption('connection')) {
             $this->connectionFactory->setConnectionName($input->getOption('connection'));
@@ -97,21 +87,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($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())];
+            $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(function (string $exception) {
-                return substr($exception, 0, 900).'…';
-            }, $job->getExceptions());
+            $exceptions = array_map(fn(string $exception) => substr($exception, 0, 900).'…', $job->getExceptions());
 
             $io->write($exceptions);
         }

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,28 +19,14 @@ 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';
-
-    /** @var ScheduledDataflowManagerInterface */
-    private $manager;
-
-    /** @var PendingDataflowRunnerInterface */
-    private $runner;
-
-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(ScheduledDataflowManagerInterface $manager, PendingDataflowRunnerInterface $runner, ConnectionFactory $connectionFactory)
+    public function __construct(private ScheduledDataflowManagerInterface $manager, private PendingDataflowRunnerInterface $runner, private ConnectionFactory $connectionFactory)
     {
         parent::__construct();
-
-        $this->manager = $manager;
-        $this->runner = $runner;
-        $this->connectionFactory = $connectionFactory;
     }
 
     /**
@@ -48,7 +35,6 @@ class RunPendingDataflowsCommand extends Command
     protected function configure()
     {
         $this
-            ->setDescription('Runs dataflows based on the scheduled defined in the UI.')
             ->setHelp(<<<'EOF'
 The <info>%command.name%</info> command runs dataflows according to the schedule defined in the UI by the user.
 EOF
@@ -59,7 +45,7 @@ EOF
     /**
      * {@inheritdoc}
      */
-    protected function execute(InputInterface $input, OutputInterface $output)
+    protected function execute(InputInterface $input, OutputInterface $output): int
     {
         if (!$this->lock()) {
             $output->writeln('The command is already running in another process.');

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,22 +16,12 @@ 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';
-
-    /** @var ScheduledDataflowRepository */
-    private $scheduledDataflowRepository;
-
-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(ScheduledDataflowRepository $scheduledDataflowRepository, ConnectionFactory $connectionFactory)
+    public function __construct(private ScheduledDataflowRepository $scheduledDataflowRepository, private ConnectionFactory $connectionFactory)
     {
         parent::__construct();
-
-        $this->scheduledDataflowRepository = $scheduledDataflowRepository;
-        $this->connectionFactory = $connectionFactory;
     }
 
     /**
@@ -39,7 +30,6 @@ class ScheduleListCommand extends Command
     protected function configure()
     {
         $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');
     }
@@ -47,7 +37,7 @@ class ScheduleListCommand extends Command
     /**
      * {@inheritdoc}
      */
-    protected function execute(InputInterface $input, OutputInterface $output)
+    protected function execute(InputInterface $input, OutputInterface $output): int
     {
         if (null !== $input->getOption('connection')) {
             $this->connectionFactory->setConnectionName($input->getOption('connection'));

src/Command/SchemaCommand.php

@@ -10,6 +10,7 @@ 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\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
@@ -19,18 +20,12 @@ use Symfony\Component\Console\Style\SymfonyStyle;
 /**
  * @codeCoverageIgnore
  */
+#[AsCommand('code-rhapsodie:dataflow:dump-schema', 'Generates schema create / update SQL queries')]
 class SchemaCommand extends Command
 {
-    protected static $defaultName = 'code-rhapsodie:dataflow:dump-schema';
-
-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(ConnectionFactory $connectionFactory)
+    public function __construct(private ConnectionFactory $connectionFactory)
     {
         parent::__construct();
-
-        $this->connectionFactory = $connectionFactory;
     }
 
     /**
@@ -39,7 +34,6 @@ class SchemaCommand extends Command
     protected function configure()
     {
         $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')
@@ -49,7 +43,7 @@ class SchemaCommand extends Command
     /**
      * {@inheritdoc}
      */
-    protected function execute(InputInterface $input, OutputInterface $output)
+    protected function execute(InputInterface $input, OutputInterface $output): int
     {
         if (null !== $input->getOption('connection')) {
             $this->connectionFactory->setConnectionName($input->getOption('connection'));

src/DataflowType/AMPAsyncDataflowBuilder.php

@@ -11,29 +11,16 @@ use Symfony\Component\OptionsResolver\OptionsResolver;
 
 class AMPAsyncDataflowBuilder extends DataflowBuilder
 {
-    /** @var int  */
-    protected $loopInterval;
-
-    /** @var int */
-    protected $emitInterval;
-
-    public function __construct(?int $loopInterval = 0, ?int $emitInterval = 0)
-    {
-        $this->loopInterval = $loopInterval;
-        $this->emitInterval = $emitInterval;
-    }
-
-    /** @var string */
-    private $name;
-
-    /** @var iterable */
-    private $reader;
-
-    /** @var array */
-    private $steps = [];
+    private ?string $name = null;
+    private ?iterable $reader = null;
+    private array $steps = [];
 
     /** @var WriterInterface[] */
-    private $writers = [];
+    private array $writers = [];
+
+    public function __construct(protected ?int $loopInterval = 0, protected ?int $emitInterval = 0)
+    {
+    }
 
     public function setName(string $name): self
     {

src/DataflowType/Dataflow/AMPAsyncDataflow.php

@@ -22,41 +22,18 @@ class AMPAsyncDataflow implements DataflowInterface, LoggerAwareInterface
 {
     use LoggerAwareTrait;
 
-    /** @var string */
-    private $name;
-
-    /** @var iterable */
-    private $reader;
-
     /** @var callable[] */
-    private $steps;
+    private array $steps = [];
 
     /** @var WriterInterface[] */
-    private $writers;
+    private array $writers = [];
 
-    /** @var int */
-    private $loopInterval;
+    private array $states = [];
 
-    /** @var int */
-    private $emitInterval;
+    private array $stepsJobs = [];
 
-    /** @var array */
-    private $states;
-
-    /** @var array */
-    private $stepsJobs;
-
-    public function __construct(iterable $reader, ?string $name, ?int $loopInterval = 0, ?int $emitInterval = 0)
+    public function __construct(private iterable $reader, private ?string $name, private ?int $loopInterval = 0, private ?int $emitInterval = 0)
     {
-        $this->reader = $reader;
-        $this->name = $name;
-        $this->steps = [];
-        $this->writers = [];
-        $this->loopInterval = $loopInterval;
-        $this->emitInterval = $emitInterval;
-        $this->states = [];
-        $this->stepsJobs = [];
-
         if (!function_exists('Amp\\Promise\\wait')) {
             throw new RuntimeException('Amp is not loaded. Suggest install it with composer require amphp/amp');
         }
@@ -91,7 +68,7 @@ class AMPAsyncDataflow implements DataflowInterface, LoggerAwareInterface
     {
         $count = 0;
         $exceptions = [];
-        $startTime = new \DateTimeImmutable();
+        $startTime = new \DateTime();
 
         try {
             foreach ($this->writers as $writer) {
@@ -133,15 +110,14 @@ 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);
     }
 
     /**
-     * @param mixed $state
      * @param int   $count      internal count reference
      * @param array $exceptions internal exceptions
      */
-    private function processState($state, int &$count, array &$exceptions): void
+    private function processState(mixed $state, int &$count, array &$exceptions): void
     {
         [$readIndex, $stepIndex, $item] = $state;
         if ($stepIndex < count($this->steps)) {
@@ -149,7 +125,7 @@ class AMPAsyncDataflow implements DataflowInterface, LoggerAwareInterface
                 $this->stepsJobs[$stepIndex] = [];
             }
             [$step, $scale] = $this->steps[$stepIndex];
-            if (count($this->stepsJobs[$stepIndex]) < $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);

src/DataflowType/Dataflow/Dataflow.php

@@ -13,24 +13,16 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface
 {
     use LoggerAwareTrait;
 
-    /** @var string */
-    private $name;
-
-    /** @var iterable */
-    private $reader;
-
     /** @var callable[] */
-    private $steps;
+    private array $steps = [];
 
     /** @var WriterInterface[] */
-    private $writers;
+    private array $writers = [];
 
-    public function __construct(iterable $reader, ?string $name)
+    private ?\Closure $customExceptionIndex = null;
+
+    public function __construct(private iterable $reader, private ?string $name)
     {
-        $this->reader = $reader;
-        $this->name = $name;
-        $this->steps = [];
-        $this->writers = [];
     }
 
     /**
@@ -53,6 +45,16 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface
         return $this;
     }
 
+    /**
+     * @return $this
+     */
+    public function setCustomExceptionIndex(callable $callable): self
+    {
+        $this->customExceptionIndex = \Closure::fromCallable($callable);
+
+        return $this;
+    }
+
     /**
      * {@inheritdoc}
      */
@@ -60,7 +62,7 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface
     {
         $count = 0;
         $exceptions = [];
-        $startTime = new \DateTimeImmutable();
+        $startTime = new \DateTime();
 
         try {
             foreach ($this->writers as $writer) {
@@ -71,8 +73,17 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface
                 try {
                     $this->processItem($item);
                 } catch (\Throwable $e) {
-                    $exceptions[$index] = $e;
-                    $this->logException($e, (string) $index);
+                    $exceptionIndex = $index;
+                    try {
+                        if (is_callable($this->customExceptionIndex)) {
+                            $exceptionIndex = (string) ($this->customExceptionIndex)($item, $index);
+                        }
+                    } catch (\Throwable $e2) {
+                        $exceptions[$index] = $e2;
+                        $this->logException($e2, $index);
+                    }
+                    $exceptions[$exceptionIndex] = $e;
+                    $this->logException($e, $exceptionIndex);
                 }
 
                 ++$count;
@@ -86,13 +97,10 @@ 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);
     }
 
-    /**
-     * @param mixed $item
-     */
-    private function processItem($item): void
+    private function processItem(mixed $item): void
     {
         foreach ($this->steps as $step) {
             $item = call_user_func($step, $item);
@@ -107,7 +115,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;

src/DataflowType/DataflowBuilder.php

@@ -10,17 +10,14 @@ use CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface;
 
 class DataflowBuilder
 {
-    /** @var string */
-    private $name;
-
-    /** @var iterable */
-    private $reader;
-
-    /** @var array */
-    private $steps = [];
+    private ?string $name = null;
+    private ?iterable $reader = null;
+    private array $steps = [];
 
     /** @var WriterInterface[] */
-    private $writers = [];
+    private array $writers = [];
+
+    private ?\Closure $customExceptionIndex = null;
 
     public function setName(string $name): self
     {
@@ -50,6 +47,13 @@ class DataflowBuilder
         return $this;
     }
 
+    public function setCustomExceptionIndex(callable $callable): self
+    {
+        $this->customExceptionIndex = \Closure::fromCallable($callable);
+
+        return $this;
+    }
+
     public function getDataflow(): DataflowInterface
     {
         $dataflow = new Dataflow($this->reader, $this->name);
@@ -65,6 +69,10 @@ class DataflowBuilder
             $dataflow->addWriter($writer);
         }
 
+        if (is_callable($this->customExceptionIndex)) {
+            $dataflow->setCustomExceptionIndex($this->customExceptionIndex);
+        }
+
         return $dataflow;
     }
 }

src/DataflowType/Result.php

@@ -9,39 +9,19 @@ namespace CodeRhapsodie\DataflowBundle\DataflowType;
  */
 class Result
 {
-    /** @var string */
-    private $name;
+    private \DateInterval $elapsed;
 
-    /** @var \DateTimeInterface */
-    private $startTime;
+    private int $errorCount = 0;
 
-    /** @var \DateTimeInterface */
-    private $endTime;
+    private int $successCount = 0;
 
-    /** @var \DateInterval */
-    private $elapsed;
+    private array $exceptions;
 
-    /** @var int */
-    private $errorCount = 0;
-
-    /** @var int */
-    private $successCount = 0;
-
-    /** @var int */
-    private $totalProcessedCount = 0;
-
-    /** @var array */
-    private $exceptions;
-
-    public function __construct(string $name, \DateTimeInterface $startTime, \DateTimeInterface $endTime, int $totalCount, array $exceptions)
+    public function __construct(private string $name, private \DateTimeInterface $startTime, private \DateTimeInterface $endTime, private int $totalProcessedCount, array $exceptions)
     {
-        $this->name = $name;
-        $this->startTime = $startTime;
-        $this->endTime = $endTime;
         $this->elapsed = $startTime->diff($endTime);
-        $this->totalProcessedCount = $totalCount;
         $this->errorCount = count($exceptions);
-        $this->successCount = $totalCount - $this->errorCount;
+        $this->successCount = $totalProcessedCount - $this->errorCount;
         $this->exceptions = $exceptions;
     }
 

src/DataflowType/Writer/CollectionWriter.php

@@ -11,15 +11,11 @@ use CodeRhapsodie\DataflowBundle\Exceptions\UnsupportedItemTypeException;
  */
 class CollectionWriter implements DelegateWriterInterface
 {
-    /** @var WriterInterface */
-    private $writer;
-
     /**
      * CollectionWriter constructor.
      */
-    public function __construct(WriterInterface $writer)
+    public function __construct(private WriterInterface $writer)
     {
-        $this->writer = $writer;
     }
 
     /**
@@ -36,7 +32,7 @@ class CollectionWriter implements DelegateWriterInterface
     public function write($collection)
     {
         if (!is_iterable($collection)) {
-            throw new UnsupportedItemTypeException(sprintf('Item to write was expected to be an iterable, received %s.', is_object($collection) ? get_class($collection) : gettype($collection)));
+            throw new UnsupportedItemTypeException(sprintf('Item to write was expected to be an iterable, received %s.', get_debug_type($collection)));
         }
 
         foreach ($collection as $item) {

src/DataflowType/Writer/DelegatorWriter.php

@@ -12,14 +12,13 @@ use CodeRhapsodie\DataflowBundle\Exceptions\UnsupportedItemTypeException;
 class DelegatorWriter implements DelegateWriterInterface
 {
     /** @var DelegateWriterInterface[] */
-    private $delegates;
+    private array $delegates = [];
 
     /**
      * DelegatorWriter constructor.
      */
     public function __construct()
     {
-        $this->delegates = [];
     }
 
     /**
@@ -47,7 +46,7 @@ class DelegatorWriter implements DelegateWriterInterface
             return;
         }
 
-        throw new UnsupportedItemTypeException(sprintf('None of the registered delegate writers support the received item of type %s', is_object($item) ? get_class($item) : gettype($item)));
+        throw new UnsupportedItemTypeException(sprintf('None of the registered delegate writers support the received item of type %s', get_debug_type($item)));
     }
 
     /**

src/DataflowType/Writer/PortWriterAdapter.php

@@ -6,12 +6,8 @@ namespace CodeRhapsodie\DataflowBundle\DataflowType\Writer;
 
 class PortWriterAdapter implements WriterInterface
 {
-    /** @var \Port\Writer */
-    private $writer;
-
-    public function __construct(\Port\Writer $writer)
+    public function __construct(private \Port\Writer $writer)
     {
-        $this->writer = $writer;
     }
 
     public function prepare()

src/DataflowType/Writer/WriterInterface.php

@@ -16,10 +16,8 @@ interface WriterInterface
 
     /**
      * Write an item.
-     *
-     * @param mixed $item
      */
-    public function write($item);
+    public function write(mixed $item);
 
     /**
      * Called after the dataflow is processed.

src/DependencyInjection/Configuration.php

@@ -10,15 +10,10 @@ use Symfony\Component\Messenger\MessageBusInterface;
 
 class Configuration implements ConfigurationInterface
 {
-    public function getConfigTreeBuilder()
+    public function getConfigTreeBuilder(): \Symfony\Component\Config\Definition\Builder\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 function ($v): bool { return $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()

src/Entity/Job.php

@@ -13,10 +13,10 @@ use Symfony\Component\Validator\Constraints as Asserts;
  */
 class Job
 {
-    const STATUS_PENDING = 0;
-    const STATUS_RUNNING = 1;
-    const STATUS_COMPLETED = 2;
-    const STATUS_QUEUED = 3;
+    public const STATUS_PENDING = 0;
+    public const STATUS_RUNNING = 1;
+    public const STATUS_COMPLETED = 2;
+    public const STATUS_QUEUED = 3;
 
     private const KEYS = [
         'id',
@@ -32,74 +32,35 @@ class Job
         'end_time',
     ];
 
-    /**
-     * @var int|null
-     */
-    private $id;
+    private ?int $id = null;
 
-    /**
-     * @var int
-     *
-     * @Asserts\Range(min=0, max=2)
-     */
-    private $status;
+    #[Asserts\Range(min: 0, max: 2)]
+    private int $status = self::STATUS_PENDING;
 
-    /**
-     * @var string|null
-     *
-     * @Asserts\NotBlank()
-     * @Asserts\Length(min=1, max=255)
-     * @Asserts\Regex("#^[[:alnum:] ]+\z#u")
-     */
-    private $label;
+    #[Asserts\NotBlank]
+    #[Asserts\Length(min: 1, max: 255)]
+    #[Asserts\Regex('#^[[:alnum:] ]+\z#u')]
+    private ?string $label = null;
 
-    /**
-     * @var string|null
-     *
-     * @Asserts\NotBlank()
-     * @Asserts\Length(min=1, max=255)
-     * @Asserts\Regex("#^[[:alnum:]\\]+\z#u")
-     */
-    private $dataflowType;
+    #[Asserts\NotBlank]
+    #[Asserts\Length(min: 1, max: 255)]
+    #[Asserts\Regex('#^[[:alnum:]\\\]+\z#u')]
+    private ?string $dataflowType = null;
 
-    /**
-     * @var array|null
-     */
-    private $options;
+    private ?array $options = null;
 
-    /**
-     * @var \DateTimeInterface|null
-     */
-    private $requestedDate;
+    private ?\DateTimeInterface $requestedDate = null;
 
-    /**
-     * @var int|null
-     */
-    private $scheduledDataflowId;
+    private ?int $scheduledDataflowId = null;
 
-    /**
-     * @var int|null
-     */
-    private $count;
+    private ?int $count = 0;
 
-    /**
-     * @var array|null
-     */
-    private $exceptions;
+    private ?array $exceptions = null;
 
-    /**
-     * @var \DateTimeInterface|null
-     */
-    private $startTime;
+    private ?\DateTimeInterface $startTime = null;
 
-    /**
-     * @var \DateTimeInterface|null
-     */
-    private $endTime;
+    private ?\DateTimeInterface $endTime = null;
 
-    /**
-     * @return Job
-     */
     public static function createFromScheduledDataflow(ScheduledDataflow $scheduled): self
     {
         return (new static())
@@ -111,12 +72,6 @@ class Job
             ->setScheduledDataflowId($scheduled->getId());
     }
 
-    public function __construct()
-    {
-        $this->count = 0;
-        $this->status = static::STATUS_PENDING;
-    }
-
     public static function createFromArray(array $datas)
     {
         $lost = array_diff(static::KEYS, array_keys($datas));

src/Entity/ScheduledDataflow.php

@@ -14,7 +14,7 @@ use Symfony\Component\Validator\Constraints as Asserts;
  */
 class ScheduledDataflow
 {
-    const AVAILABLE_FREQUENCIES = [
+    public const AVAILABLE_FREQUENCIES = [
         '1 hour',
         '1 day',
         '1 week',
@@ -23,51 +23,29 @@ class ScheduledDataflow
 
     private const KEYS = ['id', 'label', 'dataflow_type', 'options', 'frequency', 'next', 'enabled'];
 
-    /**
-     * @var int|null
-     */
-    private $id;
+    private ?int $id = null;
+
+    #[Asserts\NotBlank]
+    #[Asserts\Length(min: 1, max: 255)]
+    #[Asserts\Regex('#^[[:alnum:] ]+\z#u')]
+    private ?string $label = null;
+
+    #[Asserts\NotBlank]
+    #[Asserts\Length(min: 1, max: 255)]
+    #[Asserts\Regex('#^[[:alnum:]\\\]+\z#u')]
+    private ?string $dataflowType = null;
+
+    private ?array $options = null;
 
     /**
-     * @var string|null
-     *
-     * @Asserts\NotBlank()
-     * @Asserts\Length(min=1, max=255)
-     * @Asserts\Regex("#^[[:alnum:] ]+\z#u")
-     */
-    private $label;
-
-    /**
-     * @var string|null
-     *
-     * @Asserts\NotBlank()
-     * @Asserts\Length(min=1, max=255)
-     * @Asserts\Regex("#^[[:alnum:]\\]+\z#u")
-     */
-    private $dataflowType;
-
-    /**
-     * @var array|null
-     */
-    private $options;
-
-    /**
-     * @var string|null
-     *
-     * @Asserts\NotBlank()
      * @Frequency()
      */
-    private $frequency;
+    #[Asserts\NotBlank]
+    private ?string $frequency = null;
 
-    /**
-     * @var \DateTimeInterface|null
-     */
-    private $next;
+    private ?\DateTimeInterface $next = null;
 
-    /**
-     * @var bool|null
-     */
-    private $enabled;
+    private ?bool $enabled = null;
 
     public static function createFromArray(array $datas)
     {

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')) {
-    // 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
+{
 }

src/Event/Events.php

@@ -6,6 +6,6 @@ namespace CodeRhapsodie\DataflowBundle\Event;
 
 final class Events
 {
-    const BEFORE_PROCESSING = 'coderhapsodie.dataflow.before_processing';
-    const AFTER_PROCESSING = 'coderhapsodie.dataflow.after_processing';
+    public const BEFORE_PROCESSING = 'coderhapsodie.dataflow.before_processing';
+    public const AFTER_PROCESSING = 'coderhapsodie.dataflow.after_processing';
 }

src/Event/ProcessingEvent.php

@@ -13,15 +13,11 @@ use CodeRhapsodie\DataflowBundle\Entity\Job;
  */
 class ProcessingEvent extends CrEvent
 {
-    /** @var Job */
-    private $job;
-
     /**
      * ProcessingEvent constructor.
      */
-    public function __construct(Job $job)
+    public function __construct(private Job $job)
     {
-        $this->job = $job;
     }
 
     public function getJob(): Job

src/Factory/ConnectionFactory.php

@@ -13,14 +13,8 @@ use Symfony\Component\DependencyInjection\Container;
  */
 class ConnectionFactory
 {
-    private $connectionName;
-
-    private $container;
-
-    public function __construct(Container $container, string $connectionName)
+    public function __construct(private Container $container, private string $connectionName)
     {
-        $this->connectionName = $connectionName;
-        $this->container = $container;
     }
 
     public function setConnectionName(string $connectionName)
@@ -28,7 +22,7 @@ class ConnectionFactory
         $this->connectionName = $connectionName;
     }
 
-    public function getConnection(): \Doctrine\DBAL\Driver\Connection
+    public function getConnection(): \Doctrine\DBAL\Connection
     {
         return $this->container->get(sprintf('doctrine.dbal.%s_connection', $this->connectionName));
     }

src/Logger/BufferHandler.php

@@ -8,19 +8,13 @@ use Monolog\Formatter\FormatterInterface;
 use Monolog\Formatter\LineFormatter;
 use Monolog\Handler\AbstractProcessingHandler;
 use Monolog\Logger;
+use Monolog\LogRecord;
 
 class BufferHandler extends AbstractProcessingHandler
 {
     private const FORMAT = "[%datetime%] %level_name% when processing item %context.index%: %message% %context% %extra%\n";
 
-    private $buffer;
-
-    public function __construct($level = Logger::DEBUG, bool $bubble = true)
-    {
-        parent::__construct($level, $bubble);
-
-        $this->buffer = [];
-    }
+    private array $buffer = [];
 
     public function clearBuffer(): array
     {
@@ -30,7 +24,7 @@ class BufferHandler extends AbstractProcessingHandler
         return $logs;
     }
 
-    protected function write(array $record): void
+    protected function write(array|LogRecord $record): void
     {
         $this->buffer[] = $record['formatted'];
     }

src/Logger/DelegatingLogger.php

@@ -10,20 +10,20 @@ use Psr\Log\LoggerInterface;
 final class DelegatingLogger extends AbstractLogger
 {
     /** @var LoggerInterface[] */
-    private $loggers;
+    private ?array $loggers = null;
 
     public function __construct(iterable $loggers)
     {
         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, get_class($logger)));
+                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;
         }
     }
 
-    public function log($level, $message, array $context = [])
+    public function log($level, $message, array $context = []): void
     {
         foreach ($this->loggers as $logger) {
             $logger->log($level, $message, $context);

src/Manager/ScheduledDataflowManager.php

@@ -8,27 +8,15 @@ use CodeRhapsodie\DataflowBundle\Entity\Job;
 use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
 use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
-use Doctrine\DBAL\Driver\Connection;
+use Doctrine\DBAL\Connection;
 
 /**
  * Handles scheduled dataflows execution dates based on their frequency.
  */
 class ScheduledDataflowManager implements ScheduledDataflowManagerInterface
 {
-    /** @var ScheduledDataflowRepository */
-    private $scheduledDataflowRepository;
-
-    /** @var JobRepository */
-    private $jobRepository;
-
-    /** @var Connection */
-    private $connection;
-
-    public function __construct(Connection $connection, ScheduledDataflowRepository $scheduledDataflowRepository, JobRepository $jobRepository)
+    public function __construct(private Connection $connection, private ScheduledDataflowRepository $scheduledDataflowRepository, private JobRepository $jobRepository)
     {
-        $this->connection = $connection;
-        $this->scheduledDataflowRepository = $scheduledDataflowRepository;
-        $this->jobRepository = $jobRepository;
     }
 
     /**

src/MessengerMode/JobMessage.php

@@ -6,12 +6,8 @@ namespace CodeRhapsodie\DataflowBundle\MessengerMode;
 
 class JobMessage
 {
-    /** @var int */
-    private $jobId;
-
-    public function __construct(int $jobId)
+    public function __construct(private int $jobId)
     {
-        $this->jobId = $jobId;
     }
 
     public function getJobId(): int

src/MessengerMode/JobMessageHandler.php

@@ -6,29 +6,17 @@ 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
 {
-    /** @var JobRepository */
-    private $repository;
-
-    /** @var JobProcessorInterface */
-    private $processor;
-
-    public function __construct(JobRepository $repository, JobProcessorInterface $processor)
+    public function __construct(private JobRepository $repository, private JobProcessorInterface $processor)
     {
-        $this->repository = $repository;
-        $this->processor = $processor;
     }
 
     public function __invoke(JobMessage $message)
     {
         $this->processor->process($this->repository->find($message->getJobId()));
     }
-
-    public static function getHandledMessages(): iterable
-    {
-        return [JobMessage::class];
-    }
 }

src/Processor/JobProcessor.php

@@ -21,20 +21,8 @@ class JobProcessor implements JobProcessorInterface, LoggerAwareInterface
 {
     use LoggerAwareTrait;
 
-    /** @var JobRepository */
-    private $repository;
-
-    /** @var DataflowTypeRegistryInterface */
-    private $registry;
-
-    /** @var EventDispatcherInterface */
-    private $dispatcher;
-
-    public function __construct(JobRepository $repository, DataflowTypeRegistryInterface $registry, EventDispatcherInterface $dispatcher)
+    public function __construct(private JobRepository $repository, private DataflowTypeRegistryInterface $registry, private EventDispatcherInterface $dispatcher)
     {
-        $this->repository = $repository;
-        $this->registry = $registry;
-        $this->dispatcher = $dispatcher;
     }
 
     public function process(Job $job): void
@@ -65,12 +53,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')) {
-            $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)
@@ -89,11 +72,6 @@ class JobProcessor implements JobProcessorInterface, LoggerAwareInterface
         ;
         $this->repository->save($job);
 
-        // Symfony 3.4 to 4.4 call
-        if (!class_exists('Symfony\Contracts\EventDispatcher\Event')) {
-            $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);
     }
 }

src/Registry/DataflowTypeRegistry.php

@@ -13,10 +13,10 @@ use CodeRhapsodie\DataflowBundle\Exceptions\UnknownDataflowTypeException;
 class DataflowTypeRegistry implements DataflowTypeRegistryInterface
 {
     /** @var array|DataflowTypeInterface[] */
-    private $fqcnRegistry = [];
+    private array $fqcnRegistry = [];
 
     /** @var array|DataflowTypeInterface[] */
-    private $aliasesRegistry = [];
+    private array $aliasesRegistry = [];
 
     /**
      * {@inheritdoc}
@@ -31,7 +31,7 @@ class DataflowTypeRegistry implements DataflowTypeRegistryInterface
             return $this->aliasesRegistry[$fqcnOrAlias];
         }
 
-        throw UnknownDataflowTypeException::create($fqcnOrAlias, array_merge(array_keys($this->fqcnRegistry), array_keys($this->aliasesRegistry)));
+        throw UnknownDataflowTypeException::create($fqcnOrAlias, [...array_keys($this->fqcnRegistry), ...array_keys($this->aliasesRegistry)]);
     }
 
     /**
@@ -47,7 +47,7 @@ class DataflowTypeRegistry implements DataflowTypeRegistryInterface
      */
     public function registerDataflowType(DataflowTypeInterface $dataflowType): void
     {
-        $this->fqcnRegistry[get_class($dataflowType)] = $dataflowType;
+        $this->fqcnRegistry[$dataflowType::class] = $dataflowType;
         foreach ($dataflowType->getAliases() as $alias) {
             $this->aliasesRegistry[$alias] = $dataflowType;
         }

src/Repository/InitFromDbTrait.php

@@ -9,9 +9,11 @@ namespace CodeRhapsodie\DataflowBundle\Repository;
  */
 trait InitFromDbTrait
 {
+    abstract private function getFields(): array;
+
     private function initDateTime(array $datas): array
     {
-        foreach (static::FIELDS_TYPE as $key => $type) {
+        foreach ($this->getFields() as $key => $type) {
             if ('datetime' === $type && null !== $datas[$key]) {
                 $datas[$key] = new \DateTime($datas[$key]);
             }
@@ -38,7 +40,7 @@ trait InitFromDbTrait
             return [];
         }
 
-        $array = json_decode($value, true);
+        $array = json_decode($value, true, 512, JSON_THROW_ON_ERROR);
 
         return (false === $array) ? [] : $array;
     }

src/Repository/JobRepository.php

@@ -6,7 +6,7 @@ namespace CodeRhapsodie\DataflowBundle\Repository;
 
 use CodeRhapsodie\DataflowBundle\Entity\Job;
 use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
-use Doctrine\DBAL\Driver\Connection;
+use Doctrine\DBAL\Connection;
 use Doctrine\DBAL\ParameterType;
 use Doctrine\DBAL\Query\QueryBuilder;
 
@@ -21,31 +21,11 @@ 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',
-    ];
-
-    /**
-     * @var \Doctrine\DBAL\Connection
-     */
-    private $connection;
-
     /**
      * JobRepository constructor.
      */
-    public function __construct(Connection $connection)
+    public function __construct(private Connection $connection)
     {
-        $this->connection = $connection;
     }
 
     public function find(int $jobId)
@@ -64,7 +44,7 @@ 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();
+        $stmt = $qb->executeQuery();
         if (0 === $stmt->rowCount()) {
             return [];
         }
@@ -112,7 +92,7 @@ class JobRepository
         $qb
             ->orderBy('requested_date', 'DESC')
             ->setMaxResults(20);
-        $stmt = $qb->execute();
+        $stmt = $qb->executeQuery();
         if (0 === $stmt->rowCount()) {
             return [];
         }
@@ -127,7 +107,7 @@ class JobRepository
         $qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($id, ParameterType::INTEGER)))
             ->orderBy('requested_date', 'DESC')
             ->setMaxResults(20);
-        $stmt = $qb->execute();
+        $stmt = $qb->executeQuery();
         if (0 === $stmt->rowCount()) {
             return [];
         }
@@ -142,19 +122,19 @@ class JobRepository
         unset($datas['id']);
 
         if (is_array($datas['options'])) {
-            $datas['options'] = json_encode($datas['options']);
+            $datas['options'] = json_encode($datas['options'], JSON_THROW_ON_ERROR);
         }
         if (is_array($datas['exceptions'])) {
-            $datas['exceptions'] = json_encode($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);
+            $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 createQueryBuilder($alias = null): QueryBuilder
@@ -168,11 +148,28 @@ class JobRepository
 
     private function returnFirstOrNull(QueryBuilder $qb): ?Job
     {
-        $stmt = $qb->execute();
+        $stmt = $qb->executeQuery();
         if (0 === $stmt->rowCount()) {
             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',
+        ];
+    }
 }

src/Repository/ScheduledDataflowRepository.php

@@ -5,7 +5,7 @@ declare(strict_types=1);
 namespace CodeRhapsodie\DataflowBundle\Repository;
 
 use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
-use Doctrine\DBAL\Driver\Connection;
+use Doctrine\DBAL\Connection;
 use Doctrine\DBAL\ParameterType;
 use Doctrine\DBAL\Query\QueryBuilder;
 
@@ -20,26 +20,11 @@ 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,
-    ];
-    /**
-     * @var \Doctrine\DBAL\Connection
-     */
-    private $connection;
-
     /**
      * JobRepository constructor.
      */
-    public function __construct(Connection $connection)
+    public function __construct(private Connection $connection)
     {
-        $this->connection = $connection;
     }
 
     /**
@@ -51,11 +36,11 @@ 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();
+        $stmt = $qb->executeQuery();
         if (0 === $stmt->rowCount()) {
             return [];
         }
@@ -79,12 +64,12 @@ class ScheduledDataflowRepository
         $qb = $this->createQueryBuilder();
         $qb->orderBy('label', 'ASC');
 
-        $stmt = $qb->execute();
+        $stmt = $qb->executeQuery();
         if (0 === $stmt->rowCount()) {
             return [];
         }
         while (false !== ($row = $stmt->fetchAssociative())) {
-            yield ScheduledDataflow::createFromArray($this->initDateTime($this->initOptions($row)));
+            yield ScheduledDataflow::createFromArray($this->initDateTime($this->initArray($row)));
         }
     }
 
@@ -97,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)
@@ -106,16 +91,16 @@ class ScheduledDataflowRepository
         unset($datas['id']);
 
         if (is_array($datas['options'])) {
-            $datas['options'] = json_encode($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);
+            $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
@@ -143,11 +128,24 @@ class ScheduledDataflowRepository
 
     private function returnFirstOrNull(QueryBuilder $qb): ?ScheduledDataflow
     {
-        $stmt = $qb->execute();
+        $stmt = $qb->executeQuery();
         if (0 === $stmt->rowCount()) {
             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,
+        ];
+    }
 }

src/Runner/MessengerDataflowRunner.php

@@ -11,16 +11,8 @@ use Symfony\Component\Messenger\MessageBusInterface;
 
 class MessengerDataflowRunner implements PendingDataflowRunnerInterface
 {
-    /** @var JobRepository */
-    private $repository;
-
-    /** @var MessageBusInterface */
-    private $bus;
-
-    public function __construct(JobRepository $repository, MessageBusInterface $bus)
+    public function __construct(private JobRepository $repository, private MessageBusInterface $bus)
     {
-        $this->repository = $repository;
-        $this->bus = $bus;
     }
 
     public function runPendingDataflows(): void

src/Runner/PendingDataflowRunner.php

@@ -9,16 +9,8 @@ use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
 
 class PendingDataflowRunner implements PendingDataflowRunnerInterface
 {
-    /** @var JobRepository */
-    private $repository;
-
-    /** @var JobProcessorInterface */
-    private $processor;
-
-    public function __construct(JobRepository $repository, JobProcessorInterface $processor)
+    public function __construct(private JobRepository $repository, private JobProcessorInterface $processor)
     {
-        $this->repository = $repository;
-        $this->processor = $processor;
     }
 
     /**

src/SchemaProvider/DataflowSchemaProvider.php

@@ -47,7 +47,7 @@ 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']);
 
         return $schema;
     }

src/Validator/Constraints/FrequencyValidator.php

@@ -13,7 +13,7 @@ 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);
@@ -23,7 +23,12 @@ class FrequencyValidator extends ConstraintValidator
             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 +47,6 @@ class FrequencyValidator extends ConstraintValidator
                 ->setParameter('{{ string }}', $value)
                 ->addViolation()
             ;
-
-            return;
         }
     }
 }