Added PHP 8 support and DBAL 3 support (#59)

* Improved logging

* Better handling of default logger

* Update src/DataflowType/Dataflow/Dataflow.php

* Updated README

Co-authored-by: jbcr <51637606+jbcr@users.noreply.github.com>

.travis.yml

@@ -23,67 +23,91 @@ env:
 matrix:
   fast_finish: true
   include:
-    - php: '7.1'
-    - php: '7.2'
     - 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: '7.3'
+    - 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.1'
-      env:
-        - COMPOSER_FLAGS="--prefer-lowest"
     - 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.1'
+    - php: '7.3'
       env:
         - SYMFONY_VERSION=3.4.*
-    - php: '7.1'
+    - 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.1'
+    - php: '7.3'
       env:
         - SYMFONY_VERSION=4.3.*
-
-    # Test unsupported versions of Symfony
-    - php: '7.1'
+    - php: '7.3'
       env:
-        - SYMFONY_VERSION=4.0.*
-    - php: '7.1'
+        - SYMFONY_VERSION=5.0.*
+    - php: '7.3'
       env:
-        - SYMFONY_VERSION=4.1.*
-
+        - SYMFONY_VERSION=5.1.*
 
     # Test upcoming Symfony versions with lowest supported PHP version and dev dependencies
-    - php: '7.1'
-      env:
-        - STABILITY=dev
-        - SYMFONY_VERSION=4.4.*
+#    - php: '7.2'
+#      env:
+#        - STABILITY=dev
+#        - SYMFONY_VERSION=5.3.*
 
     # Test upcoming PHP versions with dev dependencies
-    - php: '7.4snapshot'
-      env:
-        - STABILITY=dev
-        - COMPOSER_FLAGS="--ignore-platform-reqs --prefer-stable"
+    #- 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=4.4.*
+        - SYMFONY_VERSION=5.1.*
+    - env:
+        - STABILITY=dev
+        - SYMFONY_VERSION=5.2.*
 
 before_install:
   - if [[ "$SYMFONY_VERSION" != "" ]]; then
@@ -97,15 +121,15 @@ before_install:
       phpenv config-rm xdebug.ini || true;
     fi
   - if [[ "$COVERALLS_ENABLED" == "true" ]]; then
-      travis_retry composer require --dev satooshi/php-coveralls:^2.0 --no-update $COMPOSER_FLAGS;
+      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-suggest --no-progress --ansi $COMPOSER_FLAGS
+  - 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;
+    fi;

CHANGELOG.md

@@ -1,3 +1,33 @@
+# Version 3.0.0
+
+* Added PHP 8 support
+* PHP minimum requirements bumped to 7.3
+* Added Doctrine DBAL 3 support
+* Doctrine DBAL minimum requirements bumped to 2.12
+
+# Version 2.2.0
+
+* Improve logging Dataflow job
+
+# Version 2.1.1
+
+* Fixed some Symfony 5 compatibility issues
+
+# Version 2.1.0
+
+* Added CollectionWriter and DelegatorWriter
+* Adding Symfony 5.0 compatibility
+* Save all exceptions caught in the log for `code-rhapsodie:dataflow:execute`
+* Added more output when errors occured during `code-rhapsodie:dataflow:execute`
+
+# Version 2.0.2
+
+* Fixed the connection proxy class created by the factory
+
+# Version 2.0.1
+
+* Fixed next execution time not increasing for scheduled dataflows
+
 # Version 2.0.0
 
 * Add Doctrine DBAL multi-connection support

README.md

@@ -3,7 +3,7 @@
 DataflowBundle is a bundle for Symfony 3.4+ 
 providing an easy way to create import / export dataflow.
 
-[![Build Status](https://travis-ci.org/code-rhapsodie/dataflow-bundle.svg?branch=master)](https://travis-ci.org/code-rhapsodie/dataflow-bundle)
+[![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)
 
@@ -35,12 +35,16 @@ As the following schema shows, you can define more than one dataflow:
 
 ## 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 :
 
 ```shell script
-$ composer require code-rhapsodie/dataflow
+$ composer require code-rhapsodie/dataflow-bundle
 ```
 
 #### Suggest
@@ -126,6 +130,14 @@ code_rhapsodie_dataflow:
   dbal_default_connection: test #Name of the default connection used by Dataflow bundle
 ```
 
+By default, the `logger` service will be used to log all exceptions and custom messages.
+If you want to use another logger, like a specific Monolog handler, Add this configuration:
+
+```yaml
+code_rhapsodie_dataflow:
+  default_logger: monolog.logger.custom #Service ID of the logger you want Dataflow to use
+```
+
 ## 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.
@@ -167,11 +179,11 @@ class MyFirstDataflowType extends AbstractDataflowType
 
     protected function buildDataflow(DataflowBuilder $builder, array $options): void
     {
-        
-        $this->myReader->setFilename($options['fileName']);
+        $this->myWriter->setDestinationFilePath($options['to-file']);
 
-        $builder->setReader($this->myReader)
-            ->addStep(function($data) use ($options) {
+        $builder
+            ->setReader($this->myReader->read($options['from-file']))
+            ->addStep(function ($data) use ($options) {
                 // TODO : Write your code here...
                 return $data;
             })
@@ -181,11 +193,8 @@ class MyFirstDataflowType extends AbstractDataflowType
 
     protected function configureOptions(OptionsResolver $optionsResolver): void
     {
-        $optionsResolver->setDefaults([
-            'my_option' => 'my_default_value',
-            'fileName'  => null,
-        ]);
-        $optionsResolver->setRequired('fileName');
+        $optionsResolver->setDefaults(['to-file' => '/tmp/dataflow.csv', 'from-file' => null]);
+        $optionsResolver->setRequired('from-file');
     }
 
     public function getLabel(): string
@@ -229,11 +238,8 @@ class MyFirstDataflowType extends AbstractDataflowType
     // ...
     protected function configureOptions(OptionsResolver $optionsResolver): void
     {
-        $optionsResolver->setDefaults([
-            'my_option' => 'my_default_value',
-            'fileName'  => null,
-        ]);
-        $optionsResolver->setRequired('fileName');
+        $optionsResolver->setDefaults(['to-file' => '/tmp/dataflow.csv', 'from-file' => null]);
+        $optionsResolver->setRequired('from-file');
     }
 
 }
@@ -241,6 +247,30 @@ 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).
 
+### Logging
+
+All exceptions will be caught and written in the logger.
+If you want to add custom messages in the log, you can inject the logger in your readers / steps / writers.
+If your DataflowType class extends `AbstractDataflowType`, the logger is accessible as `$this->logger`.
+
+```php
+<?php
+// ...
+use Symfony\Component\OptionsResolver\OptionsResolver;
+
+class MyDataflowType extends AbstractDataflowType
+{
+    // ...
+    protected function buildDataflow(DataflowBuilder $builder, array $options): void
+    {
+        $this->myWriter->setLogger($this->logger);
+    }
+
+}
+```
+
+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:
@@ -281,27 +311,18 @@ namespace CodeRhapsodie\DataflowExemple\Reader;
 
 class FileReader
 {
-    private $filename;
-
-    /**
-     * Set the filename option needed by the Reader.
-     */
-    public function setFilename(string $filename) {
-        $this->filename = $filename;
-    }
-
-    public function __invoke(): iterable
+    public function read(string $filename): iterable
     {
-        if (!$this->filename) {
+        if (!$filename) {
             throw new \Exception("The file name is not defined. Define it with 'setFilename' method");
         }
 
-        if (!$fh = fopen($this->filename, 'r')) {
-            throw new \Exception("Unable to open file '".$this->filename."' for read.");
+        if (!$fh = fopen($filename, 'r')) {
+            throw new \Exception("Unable to open file '".$filename."' for read.");
         }
 
-        while (false === ($read = fread($fh, 1024))) {
-            yield explode("|", $read);
+        while (false !== ($read = fgets($fh))) {
+            yield explode('|', trim($read));
         }
     }
 }
@@ -327,14 +348,16 @@ A *Step* can be any callable, taking the element as its argument, and returning
 A few examples:
 
 ```php
-$builder->addStep(function($item) {
+<?php
+//[...]
+$builder->addStep(function ($item) {
     // Titles are changed to all caps before export
     $item['title'] = strtoupper($item['title']);
 
     return $item;
 });
 
-$builder->addStep(function($item) {
+$builder->addStep(function ($item) {
     // Private items are not exported
     if ($item['private']) {
         return false;
@@ -342,6 +365,7 @@ $builder->addStep(function($item) {
 
     return $item;
 });
+//[...]
 ```
 
 ### Writers
@@ -369,11 +393,20 @@ class FileWriter implements WriterInterface
 {
     private $fh;
 
+    /** @var string */
+    private $path;
+
+    public function setDestinationFilePath(string $path) {
+        $this->path = $path;
+    }
+
     public function prepare()
     {
-
-        if (!$this->fh = fopen('/path/to/file', 'w')) {
-            throw new \Exception("Unable to open in write mode the output file.");
+        if (null === $this->path) {
+            throw new \Exception('Define the destination file name before use');
+        }
+        if (!$this->fh = fopen($this->path, 'w')) {
+            throw new \Exception('Unable to open in write mode the output file.');
         }
     }
 
@@ -389,6 +422,95 @@ 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.
+
+```php
+$builder->addWriter(new CollectionWriter($mySingleItemWriter));
+```
+
+#### DelegatorWriter
+
+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.
+
+First, create your writers implementing `DelegateWriterInterface` (this interface extends `WriterInterface` so your writers can still be used without the `DelegatorWriter`).
+
+```php
+<?php
+namespace CodeRhapsodie\DataflowExemple\Writer;
+
+use CodeRhapsodie\DataFlowBundle\DataflowType\Writer\WriterInterface;
+
+class ProductWriter implements DelegateWriterInterface
+{
+    public function supports($item): bool
+    {
+        return 'product' === reset($item);
+    }
+
+    public function prepare()
+    {
+    }
+
+    public function write($item)
+    {
+        // Process your product
+    }
+
+    public function finish()
+    {
+    }
+}
+```
+
+```php
+<?php
+namespace CodeRhapsodie\DataflowExemple\Writer;
+
+use CodeRhapsodie\DataFlowBundle\DataflowType\Writer\WriterInterface;
+
+class OrderWriter implements DelegateWriterInterface
+{
+    public function supports($item): bool
+    {
+        return 'order' === reset($item);
+    }
+
+    public function prepare()
+    {
+    }
+
+    public function write($item)
+    {
+        // Process your order
+    }
+
+    public function finish()
+    {
+    }
+}
+```
+
+Then, configure your `DelegatorWriter` and add it to your dataflow type.
+
+```php
+    protected function buildDataflow(DataflowBuilder $builder, array $options): void
+    {
+        // Snip add reader and steps
+
+        $delegatorWriter = new DelegatorWriter();
+        $delegatorWriter->addDelegate(new ProductWriter());
+        $delegatorWriter->addDelegate(new OrderWriter());
+
+        $builder->addWriter($delegatorWriter);
+    }
+```
+
+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
 
 All pending dataflow job processes are stored in a queue into the database.

Tests/DataflowType/AbstractDataflowTypeTest.php

@@ -4,7 +4,6 @@ namespace CodeRhapsodie\DataflowBundle\Tests\DataflowType;
 
 use CodeRhapsodie\DataflowBundle\DataflowType\AbstractDataflowType;
 use CodeRhapsodie\DataflowBundle\DataflowType\DataflowBuilder;
-use PHPUnit\Framework\Constraint\IsIdentical;
 use PHPUnit\Framework\TestCase;
 use Symfony\Component\OptionsResolver\OptionsResolver;
 
@@ -15,18 +14,21 @@ class AbstractDataflowTypeTest extends TestCase
         $label = 'Test label';
         $options = ['testOption' => 'Test value'];
         $values = [1, 2, 3];
+        $testCase = $this;
 
-        $dataflowType = new class($label, $options, $values) extends AbstractDataflowType
+        $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)
+            public function __construct(string $label, array $options, array $values, TestCase $testCase)
             {
                 $this->label = $label;
                 $this->options = $options;
                 $this->values = $values;
+                $this->testCase = $testCase;
             }
 
             public function getLabel(): string
@@ -42,7 +44,7 @@ class AbstractDataflowTypeTest extends TestCase
             protected function buildDataflow(DataflowBuilder $builder, array $options): void
             {
                 $builder->setReader($this->values);
-                (new IsIdentical($this->options))->evaluate($options);
+                $this->testCase->assertSame($this->options, $options);
             }
         };
 

Tests/DataflowType/Writer/CollectionWriterTest.php

@@ -0,0 +1,56 @@
+<?php
+
+namespace CodeRhapsodie\DataflowBundle\Tests\DataflowType\Writer;
+
+use CodeRhapsodie\DataflowBundle\DataflowType\Writer\CollectionWriter;
+use CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface;
+use CodeRhapsodie\DataflowBundle\Exceptions\UnsupportedItemTypeException;
+use PHPUnit\Framework\TestCase;
+
+class CollectionWriterTest extends TestCase
+{
+    public function testNotACollection()
+    {
+        $this->expectException(UnsupportedItemTypeException::class);
+
+        $writer = new CollectionWriter($this->createMock(WriterInterface::class));
+        $writer->write('Not an iterable');
+    }
+
+    public function testSupports()
+    {
+        $writer = new CollectionWriter($this->createMock(WriterInterface::class));
+
+        $this->assertTrue($writer->supports([]));
+        $this->assertTrue($writer->supports(new \ArrayIterator([])));
+        $this->assertFalse($writer->supports(''));
+        $this->assertFalse($writer->supports(0));
+    }
+
+    public function testAll()
+    {
+        $values = ['a', 'b', 'c'];
+
+        $embeddedWriter = $this->createMock(WriterInterface::class);
+        $embeddedWriter
+            ->expects($this->once())
+            ->method('prepare')
+        ;
+        $embeddedWriter
+            ->expects($this->once())
+            ->method('finish')
+        ;
+        $embeddedWriter
+            ->expects($this->exactly(count($values)))
+            ->method('write')
+            ->withConsecutive(...array_map(function ($item) {
+                return [$item];
+            }, $values))
+        ;
+
+        $writer = new CollectionWriter($embeddedWriter);
+        $writer->prepare();
+        $writer->write($values);
+        $writer->finish();
+    }
+}

Tests/DataflowType/Writer/DelegatorWriterTest.php

@@ -0,0 +1,160 @@
+<?php
+
+namespace CodeRhapsodie\DataflowBundle\Tests\DataflowType\Writer;
+
+use CodeRhapsodie\DataflowBundle\DataflowType\Writer\DelegateWriterInterface;
+use CodeRhapsodie\DataflowBundle\DataflowType\Writer\DelegatorWriter;
+use CodeRhapsodie\DataflowBundle\Exceptions\UnsupportedItemTypeException;
+use PHPUnit\Framework\MockObject\MockObject;
+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;
+
+    protected function setUp(): void
+    {
+        $this->delegateInt = $this->createMock(DelegateWriterInterface::class);
+        $this->delegateInt->method('supports')->willReturnCallback(function ($argument) {
+            return is_int($argument);
+        });
+
+        $this->delegateString = $this->createMock(DelegateWriterInterface::class);
+        $this->delegateString->method('supports')->willReturnCallback(function ($argument) {
+            return is_string($argument);
+        });
+
+        $this->delegateArray = $this->createMock(DelegateWriterInterface::class);
+        $this->delegateArray->method('supports')->willReturnCallback(function ($argument) {
+            return is_array($argument);
+        });
+
+        $this->delegatorWriter = new DelegatorWriter();
+        $this->delegatorWriter->addDelegates([
+            $this->delegateInt,
+            $this->delegateString,
+            $this->delegateArray,
+        ]);
+    }
+
+    public function testUnsupported()
+    {
+        $this->expectException(UnsupportedItemTypeException::class);
+
+        $this->delegatorWriter->write(new \stdClass());
+    }
+
+    public function testStopAtFirstSupportingDelegate()
+    {
+        $value = 0;
+
+        $this->delegateInt->expects($this->once())->method('supports');
+        $this->delegateInt
+            ->expects($this->once())
+            ->method('write')
+            ->with($value)
+        ;
+        $this->delegateString->expects($this->never())->method('supports');
+        $this->delegateArray->expects($this->never())->method('supports');
+        $this->delegateString->expects($this->never())->method('write');
+        $this->delegateArray->expects($this->never())->method('write');
+
+        $this->delegatorWriter->write($value);
+    }
+
+    public function testNotSupported()
+    {
+        $value = new \stdClass();
+
+        $this->delegateInt
+            ->expects($this->once())
+            ->method('supports')
+            ->with($value)
+        ;
+        $this->delegateString
+            ->expects($this->once())
+            ->method('supports')
+            ->with($value)
+        ;
+        $this->delegateArray
+            ->expects($this->once())
+            ->method('supports')
+            ->with($value)
+        ;
+
+        $this->assertFalse($this->delegatorWriter->supports($value));
+    }
+
+    public function testSupported()
+    {
+        $value = '';
+
+        $this->delegateInt
+            ->expects($this->once())
+            ->method('supports')
+            ->with($value)
+        ;
+        $this->delegateString
+            ->expects($this->once())
+            ->method('supports')
+            ->with($value)
+        ;
+        $this->delegateArray
+            ->expects($this->never())
+            ->method('supports')
+        ;
+
+        $this->assertTrue($this->delegatorWriter->supports($value));
+    }
+
+    public function testAll()
+    {
+        $value = ['a'];
+
+        $this->delegateInt
+            ->expects($this->once())
+            ->method('supports')
+            ->with($value)
+        ;
+        $this->delegateString
+            ->expects($this->once())
+            ->method('supports')
+            ->with($value)
+        ;
+        $this->delegateArray
+            ->expects($this->once())
+            ->method('supports')
+            ->with($value)
+        ;
+
+        $this->delegateInt->expects($this->once())->method('prepare');
+        $this->delegateString->expects($this->once())->method('prepare');
+        $this->delegateArray->expects($this->once())->method('prepare');
+
+        $this->delegateInt->expects($this->once())->method('finish');
+        $this->delegateString->expects($this->once())->method('finish');
+        $this->delegateArray->expects($this->once())->method('finish');
+
+        $this->delegateInt->expects($this->never())->method('write');
+        $this->delegateString->expects($this->never())->method('write');
+        $this->delegateArray
+            ->expects($this->once())
+            ->method('write')
+            ->with($value)
+        ;
+
+        $this->delegatorWriter->prepare();
+        $this->delegatorWriter->write($value);
+        $this->delegatorWriter->finish();
+    }
+}

Tests/Manager/ScheduledDataflowManagerTest.php

@@ -83,6 +83,12 @@ class ScheduledDataflowManagerTest extends TestCase
             )
         ;
 
+        $this->scheduledDataflowRepository
+            ->expects($this->once())
+            ->method('save')
+            ->with($scheduled2)
+        ;
+
         $this->connection
             ->expects($this->once())
             ->method('commit')
@@ -92,4 +98,44 @@ class ScheduledDataflowManagerTest extends TestCase
 
         $this->assertEquals($next->add(\DateInterval::createFromDateString($frequency)), $scheduled2->getNext());
     }
+
+    public function testCreateJobsFromScheduledDataflowsWithError()
+    {
+        $scheduled1 = new ScheduledDataflow();
+
+        $this->scheduledDataflowRepository
+            ->expects($this->once())
+            ->method('findReadyToRun')
+            ->willReturn([$scheduled1])
+        ;
+
+        $this->jobRepository
+            ->expects($this->exactly(1))
+            ->method('findPendingForScheduledDataflow')
+            ->withConsecutive([$scheduled1])
+            ->willThrowException(new \Exception())
+        ;
+
+        $this->connection
+            ->expects($this->once())
+            ->method('beginTransaction')
+        ;
+        $this->jobRepository
+            ->expects($this->never())
+            ->method('save')
+        ;
+
+        $this->connection
+            ->expects($this->never())
+            ->method('commit')
+        ;
+        $this->connection
+            ->expects($this->once())
+            ->method('rollBack')
+        ;
+
+        $this->expectException(\Exception::class);
+
+        $this->manager->createJobsFromScheduledDataflows();
+    }
 }

Tests/Runner/PendingDataflowRunnerTest.php

@@ -58,36 +58,68 @@ class PendingDataflowRunnerTest extends TestCase
             ->willReturnOnConsecutiveCalls($job1, $job2, null)
         ;
 
-        $this->dispatcher
-            ->expects($this->exactly(4))
-            ->method('dispatch')
-            ->withConsecutive(
-                [
-                    Events::BEFORE_PROCESSING,
-                    $this->callback(function (ProcessingEvent $event) use ($job1) {
-                        return $event->getJob() === $job1;
-                    })
-                ],
-                [
-                    Events::AFTER_PROCESSING,
-                    $this->callback(function (ProcessingEvent $event) use ($job1) {
-                        return $event->getJob() === $job1;
-                    })
-                ],
-                [
-                    Events::BEFORE_PROCESSING,
-                    $this->callback(function (ProcessingEvent $event) use ($job2) {
-                        return $event->getJob() === $job2;
-                    })
-                ],
-                [
-                    Events::AFTER_PROCESSING,
-                    $this->callback(function (ProcessingEvent $event) use ($job2) {
-                        return $event->getJob() === $job2;
-                    })
-                ]
-            )
-        ;
+        // Symfony 3.4 to 4.4 call
+        if (!class_exists('Symfony\Contracts\EventDispatcher\Event')) {
+            $this->dispatcher
+                ->expects($this->exactly(4))
+                ->method('dispatch')
+                ->withConsecutive(
+                    [
+                        Events::BEFORE_PROCESSING,
+                        $this->callback(function (ProcessingEvent $event) use ($job1) {
+                            return $event->getJob() === $job1;
+                        })
+                    ],
+                    [
+                        Events::AFTER_PROCESSING,
+                        $this->callback(function (ProcessingEvent $event) use ($job1) {
+                            return $event->getJob() === $job1;
+                        })
+                    ],
+                    [
+                        Events::BEFORE_PROCESSING,
+                        $this->callback(function (ProcessingEvent $event) use ($job2) {
+                            return $event->getJob() === $job2;
+                        })
+                    ],
+                    [
+                        Events::AFTER_PROCESSING,
+                        $this->callback(function (ProcessingEvent $event) use ($job2) {
+                            return $event->getJob() === $job2;
+                        })
+                    ]
+                );
+        } else { // Symfony 5.0+
+            $this->dispatcher
+                ->expects($this->exactly(4))
+                ->method('dispatch')
+                ->withConsecutive(
+                    [
+                        $this->callback(function (ProcessingEvent $event) use ($job1) {
+                            return $event->getJob() === $job1;
+                        }),
+                        Events::BEFORE_PROCESSING,
+                    ],
+                    [
+                        $this->callback(function (ProcessingEvent $event) use ($job1) {
+                            return $event->getJob() === $job1;
+                        }),
+                        Events::AFTER_PROCESSING,
+                    ],
+                    [
+                        $this->callback(function (ProcessingEvent $event) use ($job2) {
+                            return $event->getJob() === $job2;
+                        }),
+                        Events::BEFORE_PROCESSING,
+                    ],
+                    [
+                        $this->callback(function (ProcessingEvent $event) use ($job2) {
+                            return $event->getJob() === $job2;
+                        }),
+                        Events::AFTER_PROCESSING,
+                    ]
+                );
+        }
 
         $dataflowType1 = $this->createMock(DataflowTypeInterface::class);
         $dataflowType2 = $this->createMock(DataflowTypeInterface::class);

composer.json

@@ -2,7 +2,12 @@
     "name": "code-rhapsodie/dataflow-bundle",
     "description": "Data processing framework inspired by PortPHP",
     "type": "symfony-bundle",
-    "keywords": ["dataflow", "import", "export", "data processing"],
+    "keywords": [
+        "dataflow",
+        "import",
+        "export",
+        "data processing"
+    ],
     "license": "MIT",
     "authors": [
         {
@@ -36,22 +41,23 @@
         }
     },
     "require": {
-        "php": "^7.1",
-        "doctrine/dbal": "^2.0",
-        "seld/signal-handler": "^1.0",
-        "symfony/config": "^3.4||^4.0",
-        "symfony/console": "^3.4||^4.0",
-        "symfony/dependency-injection": "^3.4||^4.0",
-        "symfony/event-dispatcher": "^3.4||^4.0",
-        "symfony/http-kernel": "^3.4||^4.0",
-        "symfony/lock": "^3.4||^4.0",
-        "symfony/options-resolver": "^3.4||^4.0",
-        "symfony/validator": "^3.4||^4.0",
-        "symfony/yaml": "^3.4||^4.0",
-        "doctrine/doctrine-bundle": "^1.0"
+        "php": "^7.3||^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"
     },
     "require-dev": {
-        "friendsofphp/php-cs-fixer": "^2.15",
         "phpunit/phpunit": "^7||^8"
     },
     "suggest": {
@@ -62,7 +68,8 @@
     },
     "extra": {
         "branch-alias": {
-            "dev-master": "2.x-dev",
+            "dev-master": "2.1.x-dev",
+            "dev-v2.0.x": "2.0.x-dev",
             "dev-v1.x": "1.x-dev"
         }
     }

src/CodeRhapsodieDataflowBundle.php

@@ -6,6 +6,7 @@ namespace CodeRhapsodie\DataflowBundle;
 
 use CodeRhapsodie\DataflowBundle\DependencyInjection\CodeRhapsodieDataflowExtension;
 use CodeRhapsodie\DataflowBundle\DependencyInjection\Compiler\DataflowTypeCompilerPass;
+use CodeRhapsodie\DataflowBundle\DependencyInjection\Compiler\DefaultLoggerCompilerPass;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\HttpKernel\Bundle\Bundle;
 
@@ -23,6 +24,9 @@ class CodeRhapsodieDataflowBundle extends Bundle
 
     public function build(ContainerBuilder $container)
     {
-        $container->addCompilerPass(new DataflowTypeCompilerPass());
+        $container
+            ->addCompilerPass(new DataflowTypeCompilerPass())
+            ->addCompilerPass(new DefaultLoggerCompilerPass())
+        ;
     }
 }

src/Command/AddScheduledDataflowCommand.php

@@ -5,6 +5,7 @@ declare(strict_types=1);
 namespace CodeRhapsodie\DataflowBundle\Command;
 
 use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
+use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
 use Symfony\Component\Console\Command\Command;
@@ -13,7 +14,6 @@ use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Console\Style\SymfonyStyle;
 use Symfony\Component\Validator\Validator\ValidatorInterface;
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 
 /**
  * @codeCoverageIgnore

src/Command/ChangeScheduleStatusCommand.php

@@ -5,6 +5,7 @@ declare(strict_types=1);
 namespace CodeRhapsodie\DataflowBundle\Command;
 
 use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
+use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
@@ -12,7 +13,6 @@ use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Console\Style\SymfonyStyle;
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 
 /**
  * @codeCoverageIgnore

src/Command/ExecuteDataflowCommand.php

@@ -4,21 +4,27 @@ declare(strict_types=1);
 
 namespace CodeRhapsodie\DataflowBundle\Command;
 
+use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
+use Psr\Log\LoggerAwareInterface;
+use Psr\Log\LoggerAwareTrait;
+use Psr\Log\LoggerInterface;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
+use Symfony\Component\Console\Style\SymfonyStyle;
 
 /**
  * Runs one dataflow.
  *
  * @codeCoverageIgnore
  */
-class ExecuteDataflowCommand extends Command
+class ExecuteDataflowCommand extends Command implements LoggerAwareInterface
 {
+    use LoggerAwareTrait;
+
     protected static $defaultName = 'code-rhapsodie:dataflow:execute';
 
     /** @var DataflowTypeRegistryInterface */
@@ -63,14 +69,25 @@ EOF
         }
         $fqcnOrAlias = $input->getArgument('fqcn');
         $options = json_decode($input->getArgument('options'), true);
+        $io = new SymfonyStyle($input, $output);
 
         $dataflowType = $this->registry->getDataflowType($fqcnOrAlias);
+        if ($dataflowType instanceof LoggerAwareInterface && isset($this->logger)) {
+            $dataflowType->setLogger($this->logger);
+        }
+
         $result = $dataflowType->process($options);
 
-        $output->writeln('Executed: '.$result->getName());
-        $output->writeln('Start time: '.$result->getStartTime()->format('Y/m/d H:i:s'));
-        $output->writeln('End time: '.$result->getEndTime()->format('Y/m/d H:i:s'));
-        $output->writeln('Success: '.$result->getSuccessCount());
+        $io->writeln('Executed: '.$result->getName());
+        $io->writeln('Start time: '.$result->getStartTime()->format('Y/m/d H:i:s'));
+        $io->writeln('End time: '.$result->getEndTime()->format('Y/m/d H:i:s'));
+        $io->writeln('Success: '.$result->getSuccessCount());
+
+        if ($result->hasErrors()) {
+            $io->error("Errors: {$result->getErrorCount()}\nExceptions traces are available in the logs.");
+
+            return 1;
+        }
 
         return 0;
     }

src/Command/JobShowCommand.php

@@ -5,13 +5,13 @@ declare(strict_types=1);
 namespace CodeRhapsodie\DataflowBundle\Command;
 
 use CodeRhapsodie\DataflowBundle\Entity\Job;
+use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Console\Style\SymfonyStyle;
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 
 /**
  * @codeCoverageIgnore

src/Command/RunPendingDataflowsCommand.php

@@ -4,6 +4,7 @@ declare(strict_types=1);
 
 namespace CodeRhapsodie\DataflowBundle\Command;
 
+use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Manager\ScheduledDataflowManagerInterface;
 use CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunnerInterface;
 use Symfony\Component\Console\Command\Command;
@@ -11,7 +12,6 @@ use Symfony\Component\Console\Command\LockableTrait;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 
 /**
  * Runs dataflows according to user-defined schedule.

src/Command/ScheduleListCommand.php

@@ -4,13 +4,13 @@ declare(strict_types=1);
 
 namespace CodeRhapsodie\DataflowBundle\Command;
 
+use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Console\Style\SymfonyStyle;
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 
 /**
  * @codeCoverageIgnore

src/Command/SchemaCommand.php

@@ -4,6 +4,7 @@ declare(strict_types=1);
 
 namespace CodeRhapsodie\DataflowBundle\Command;
 
+use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
 use CodeRhapsodie\DataflowBundle\SchemaProvider\DataflowSchemaProvider;
@@ -14,7 +15,6 @@ use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Console\Style\SymfonyStyle;
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 
 /**
  * @codeCoverageIgnore
@@ -93,7 +93,9 @@ class SchemaCommand extends Command
         $io = new SymfonyStyle($input, $output);
         $io->text('Execute these SQL Queries on your database:');
         foreach ($sqls as $sql) {
-            $io->text($sql);
+            $io->text($sql.';');
         }
+
+        return 0;
     }
 }

src/DataflowType/AbstractDataflowType.php

@@ -4,10 +4,15 @@ declare(strict_types=1);
 
 namespace CodeRhapsodie\DataflowBundle\DataflowType;
 
+use Psr\Log\LoggerAwareInterface;
+use Psr\Log\LoggerAwareTrait;
+use Psr\Log\LoggerInterface;
 use Symfony\Component\OptionsResolver\OptionsResolver;
 
-abstract class AbstractDataflowType implements DataflowTypeInterface
+abstract class AbstractDataflowType implements DataflowTypeInterface, LoggerAwareInterface
 {
+    use LoggerAwareTrait;
+
     /**
      * @codeCoverageIgnore
      */
@@ -27,6 +32,9 @@ abstract class AbstractDataflowType implements DataflowTypeInterface
         ;
         $this->buildDataflow($builder, $options);
         $dataflow = $builder->getDataflow();
+        if ($dataflow instanceof LoggerAwareInterface && $this->logger instanceof LoggerInterface) {
+            $dataflow->setLogger($this->logger);
+        }
 
         return $dataflow->process();
     }

src/DataflowType/Dataflow/Dataflow.php

@@ -6,9 +6,13 @@ namespace CodeRhapsodie\DataflowBundle\DataflowType\Dataflow;
 
 use CodeRhapsodie\DataflowBundle\DataflowType\Result;
 use CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface;
+use Psr\Log\LoggerAwareInterface;
+use Psr\Log\LoggerAwareTrait;
 
-class Dataflow implements DataflowInterface
+class Dataflow implements DataflowInterface, LoggerAwareInterface
 {
+    use LoggerAwareTrait;
+
     /** @var string */
     private $name;
 
@@ -16,24 +20,20 @@ class Dataflow implements DataflowInterface
     private $reader;
 
     /** @var callable[] */
-    private $steps = [];
+    private $steps;
 
     /** @var WriterInterface[] */
-    private $writers = [];
+    private $writers;
 
-    /**
-     * @param iterable    $reader
-     * @param string|null $name
-     */
     public function __construct(iterable $reader, ?string $name)
     {
         $this->reader = $reader;
         $this->name = $name;
+        $this->steps = [];
+        $this->writers = [];
     }
 
     /**
-     * @param callable $step
-     *
      * @return $this
      */
     public function addStep(callable $step): self
@@ -44,8 +44,6 @@ class Dataflow implements DataflowInterface
     }
 
     /**
-     * @param WriterInterface $writer
-     *
      * @return $this
      */
     public function addWriter(WriterInterface $writer): self
@@ -62,27 +60,33 @@ class Dataflow implements DataflowInterface
     {
         $count = 0;
         $exceptions = [];
-        $startTime = new \DateTime();
+        $startTime = new \DateTimeImmutable();
 
-        foreach ($this->writers as $writer) {
-            $writer->prepare();
-        }
-
-        foreach ($this->reader as $index => $item) {
-            try {
-                $this->processItem($item);
-            } catch (\Exception $e) {
-                $exceptions[$index] = $e;
+        try {
+            foreach ($this->writers as $writer) {
+                $writer->prepare();
             }
 
-            ++$count;
+            foreach ($this->reader as $index => $item) {
+                try {
+                    $this->processItem($item);
+                } catch (\Throwable $e) {
+                    $exceptions[$index] = $e;
+                    $this->logException($e, (string) $index);
+                }
+
+                ++$count;
+            }
+
+            foreach ($this->writers as $writer) {
+                $writer->finish();
+            }
+        } catch (\Throwable $e) {
+            $exceptions[] = $e;
+            $this->logException($e);
         }
 
-        foreach ($this->writers as $writer) {
-            $writer->finish();
-        }
-
-        return new Result($this->name, $startTime, new \DateTime(), $count, $exceptions);
+        return new Result($this->name, $startTime, new \DateTimeImmutable(), $count, $exceptions);
     }
 
     /**
@@ -102,4 +106,13 @@ class Dataflow implements DataflowInterface
             $writer->write($item);
         }
     }
+
+    private function logException(\Throwable $e, ?string $index = null): void
+    {
+        if (!isset($this->logger)) {
+            return;
+        }
+
+        $this->logger->error($e, ['exception' => $e, 'index' => $index]);
+    }
 }

src/DataflowType/Dataflow/DataflowInterface.php

@@ -13,8 +13,6 @@ interface DataflowInterface
 {
     /**
      * Processes the data.
-     *
-     * @return Result
      */
     public function process(): Result;
 }

src/DataflowType/Result.php

@@ -33,13 +33,6 @@ class Result
     /** @var array */
     private $exceptions;
 
-    /**
-     * @param string             $name
-     * @param \DateTimeInterface $startTime
-     * @param \DateTimeInterface $endTime
-     * @param int                $totalCount
-     * @param \SplObjectStorage  $exceptions
-     */
     public function __construct(string $name, \DateTimeInterface $startTime, \DateTimeInterface $endTime, int $totalCount, array $exceptions)
     {
         $this->name = $name;
@@ -52,73 +45,46 @@ class Result
         $this->exceptions = $exceptions;
     }
 
-    /**
-     * @return string
-     */
     public function getName(): string
     {
         return $this->name;
     }
 
-    /**
-     * @return \DateTimeInterface
-     */
     public function getStartTime(): \DateTimeInterface
     {
         return $this->startTime;
     }
 
-    /**
-     * @return \DateTimeInterface
-     */
     public function getEndTime(): \DateTimeInterface
     {
         return $this->endTime;
     }
 
-    /**
-     * @return \DateInterval
-     */
     public function getElapsed(): \DateInterval
     {
         return $this->elapsed;
     }
 
-    /**
-     * @return int
-     */
     public function getErrorCount(): int
     {
         return $this->errorCount;
     }
 
-    /**
-     * @return int
-     */
     public function getSuccessCount(): int
     {
         return $this->successCount;
     }
 
-    /**
-     * @return int
-     */
     public function getTotalProcessedCount(): int
     {
         return $this->totalProcessedCount;
     }
 
-    /**
-     * @return bool
-     */
     public function hasErrors(): bool
     {
         return $this->errorCount > 0;
     }
 
-    /**
-     * @return array
-     */
     public function getExceptions(): array
     {
         return $this->exceptions;

src/DataflowType/Writer/CollectionWriter.php

@@ -0,0 +1,62 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\DataflowType\Writer;
+
+use CodeRhapsodie\DataflowBundle\Exceptions\UnsupportedItemTypeException;
+
+/**
+ * Delegates the writing of each item in a collection to an embedded writer.
+ */
+class CollectionWriter implements DelegateWriterInterface
+{
+    /** @var WriterInterface */
+    private $writer;
+
+    /**
+     * CollectionWriter constructor.
+     */
+    public function __construct(WriterInterface $writer)
+    {
+        $this->writer = $writer;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function prepare()
+    {
+        $this->writer->prepare();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function write($collection)
+    {
+        if (!is_iterable($collection)) {
+            throw new UnsupportedItemTypeException(sprintf('Item to write was expected to be an iterable, received %s.', is_object($collection) ? get_class($collection) : gettype($collection)));
+        }
+
+        foreach ($collection as $item) {
+            $this->writer->write($item);
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function finish()
+    {
+        $this->writer->finish();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function supports($item): bool
+    {
+        return is_iterable($item);
+    }
+}

src/DataflowType/Writer/DelegateWriterInterface.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\DataflowType\Writer;
+
+/**
+ * A writer that can be used as a delegate of DelegatorWriter.
+ */
+interface DelegateWriterInterface extends WriterInterface
+{
+    /**
+     * Returns true if the argument is of a supported type.
+     *
+     * @param $item
+     */
+    public function supports($item): bool;
+}

src/DataflowType/Writer/DelegatorWriter.php

@@ -0,0 +1,96 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\DataflowType\Writer;
+
+use CodeRhapsodie\DataflowBundle\Exceptions\UnsupportedItemTypeException;
+
+/**
+ * Writer that delegated the actual writing to other writers.
+ */
+class DelegatorWriter implements DelegateWriterInterface
+{
+    /** @var DelegateWriterInterface[] */
+    private $delegates;
+
+    /**
+     * DelegatorWriter constructor.
+     */
+    public function __construct()
+    {
+        $this->delegates = [];
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function prepare()
+    {
+        foreach ($this->delegates as $delegate) {
+            $delegate->prepare();
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function write($item)
+    {
+        foreach ($this->delegates as $delegate) {
+            if (!$delegate->supports($item)) {
+                continue;
+            }
+
+            $delegate->write($item);
+
+            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)));
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function finish()
+    {
+        foreach ($this->delegates as $delegate) {
+            $delegate->finish();
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function supports($item): bool
+    {
+        foreach ($this->delegates as $delegate) {
+            if ($delegate->supports($item)) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    /**
+     * Registers a collection of delegates.
+     *
+     * @param iterable|DelegateWriterInterface[] $delegates
+     */
+    public function addDelegates(iterable $delegates): void
+    {
+        foreach ($delegates as $delegate) {
+            $this->addDelegate($delegate);
+        }
+    }
+
+    /**
+     * Registers one delegate.
+     */
+    public function addDelegate(DelegateWriterInterface $delegate): void
+    {
+        $this->delegates[] = $delegate;
+    }
+}

src/DataflowType/Writer/WriterInterface.php

@@ -4,11 +4,25 @@ declare(strict_types=1);
 
 namespace CodeRhapsodie\DataflowBundle\DataflowType\Writer;
 
+/**
+ * Represents a writer for dataflows.
+ */
 interface WriterInterface
 {
+    /**
+     * Called before the dataflow is processed.
+     */
     public function prepare();
 
+    /**
+     * Write an item.
+     *
+     * @param mixed $item
+     */
     public function write($item);
 
+    /**
+     * Called after the dataflow is processed.
+     */
     public function finish();
 }

src/DependencyInjection/CodeRhapsodieDataflowExtension.php

@@ -28,5 +28,6 @@ class CodeRhapsodieDataflowExtension extends Extension
         $config = $this->processConfiguration($configuration, $configs);
 
         $container->setParameter('coderhapsodie.dataflow.dbal_default_connection', $config['dbal_default_connection']);
+        $container->setParameter('coderhapsodie.dataflow.default_logger', $config['default_logger']);
     }
 }

src/DependencyInjection/Compiler/DefaultLoggerCompilerPass.php

@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\DependencyInjection\Compiler;
+
+use CodeRhapsodie\DataflowBundle\Command\ExecuteDataflowCommand;
+use CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunner;
+use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\DependencyInjection\Reference;
+
+class DefaultLoggerCompilerPass implements CompilerPassInterface
+{
+    /**
+     * {@inheritdoc}
+     */
+    public function process(ContainerBuilder $container)
+    {
+        $defaultLogger = $container->getParameter('coderhapsodie.dataflow.default_logger');
+        if (!$container->has($defaultLogger)) {
+            return;
+        }
+
+        foreach ([ExecuteDataflowCommand::class, PendingDataflowRunner::class] as $serviceId) {
+            if (!$container->has($serviceId)) {
+                continue;
+            }
+
+            $definition = $container->findDefinition($serviceId);
+            $definition->addMethodCall('setLogger', [new Reference($defaultLogger)]);
+        }
+    }
+}

src/DependencyInjection/Configuration.php

@@ -11,14 +11,22 @@ class Configuration implements ConfigurationInterface
 {
     public function getConfigTreeBuilder()
     {
-        $treeBuilder = new TreeBuilder();
-        $rootNode = $treeBuilder->root('code_rhapsodie_dataflow');
+        $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
             ->children()
                 ->scalarNode('dbal_default_connection')
                     ->defaultValue('default')
                 ->end()
+                ->scalarNode('default_logger')
+                    ->defaultValue('logger')
+                ->end()
             ->end()
         ;
 

src/Entity/Job.php

@@ -68,8 +68,6 @@ class Job
 
     /**
      * @var \DateTimeInterface|null
-     *
-     * @Asserts\DateTime()
      */
     private $requestedDate;
 
@@ -99,8 +97,6 @@ class Job
     private $endTime;
 
     /**
-     * @param ScheduledDataflow $scheduled
-     *
      * @return Job
      */
     public static function createFromScheduledDataflow(ScheduledDataflow $scheduled): self
@@ -124,8 +120,7 @@ class Job
     {
         $lost = array_diff(static::KEYS, array_keys($datas));
         if (count($lost) > 0) {
-            throw new \LogicException('The first argument of '.__METHOD__.'  must be contains: "'.implode(', ',
-                    $lost).'"');
+            throw new \LogicException('The first argument of '.__METHOD__.'  must be contains: "'.implode(', ', $lost).'"');
         }
 
         $job = new self();
@@ -161,11 +156,6 @@ class Job
         ];
     }
 
-    /**
-     * @param int $id
-     *
-     * @return Job
-     */
     public function setId(int $id): Job
     {
         $this->id = $id;
@@ -173,27 +163,16 @@ class Job
         return $this;
     }
 
-    /**
-     * @return int|null
-     */
     public function getId(): ?int
     {
         return $this->id;
     }
 
-    /**
-     * @return int
-     */
     public function getStatus(): int
     {
         return $this->status;
     }
 
-    /**
-     * @param int $status
-     *
-     * @return Job
-     */
     public function setStatus(int $status): Job
     {
         $this->status = $status;
@@ -201,19 +180,11 @@ class Job
         return $this;
     }
 
-    /**
-     * @return string|null
-     */
     public function getLabel(): ?string
     {
         return $this->label;
     }
 
-    /**
-     * @param string|null $label
-     *
-     * @return Job
-     */
     public function setLabel(?string $label): Job
     {
         $this->label = $label;
@@ -221,19 +192,11 @@ class Job
         return $this;
     }
 
-    /**
-     * @return string|null
-     */
     public function getDataflowType(): ?string
     {
         return $this->dataflowType;
     }
 
-    /**
-     * @param string|null $dataflowType
-     *
-     * @return Job
-     */
     public function setDataflowType(?string $dataflowType): Job
     {
         $this->dataflowType = $dataflowType;
@@ -241,19 +204,11 @@ class Job
         return $this;
     }
 
-    /**
-     * @return array|null
-     */
     public function getOptions(): ?array
     {
         return $this->options;
     }
 
-    /**
-     * @param array|null $options
-     *
-     * @return Job
-     */
     public function setOptions(?array $options): Job
     {
         $this->options = $options;
@@ -261,19 +216,11 @@ class Job
         return $this;
     }
 
-    /**
-     * @return \DateTimeInterface|null
-     */
     public function getRequestedDate(): ?\DateTimeInterface
     {
         return $this->requestedDate;
     }
 
-    /**
-     * @param \DateTimeInterface|null $requestedDate
-     *
-     * @return Job
-     */
     public function setRequestedDate(?\DateTimeInterface $requestedDate): Job
     {
         $this->requestedDate = $requestedDate;
@@ -281,19 +228,11 @@ class Job
         return $this;
     }
 
-    /**
-     * @return int|null
-     */
     public function getScheduledDataflowId(): ?int
     {
         return $this->scheduledDataflowId;
     }
 
-    /**
-     * @param int|null $scheduledDataflowId
-     *
-     * @return Job
-     */
     public function setScheduledDataflowId(?int $scheduledDataflowId): Job
     {
         $this->scheduledDataflowId = $scheduledDataflowId;
@@ -301,19 +240,11 @@ class Job
         return $this;
     }
 
-    /**
-     * @return int|null
-     */
     public function getCount(): ?int
     {
         return $this->count;
     }
 
-    /**
-     * @param int|null $count
-     *
-     * @return Job
-     */
     public function setCount(?int $count): Job
     {
         $this->count = $count;
@@ -321,19 +252,11 @@ class Job
         return $this;
     }
 
-    /**
-     * @return array|null
-     */
     public function getExceptions(): ?array
     {
         return $this->exceptions;
     }
 
-    /**
-     * @param array|null $exceptions
-     *
-     * @return Job
-     */
     public function setExceptions(?array $exceptions): Job
     {
         $this->exceptions = $exceptions;
@@ -341,19 +264,11 @@ class Job
         return $this;
     }
 
-    /**
-     * @return \DateTimeInterface|null
-     */
     public function getStartTime(): ?\DateTimeInterface
     {
         return $this->startTime;
     }
 
-    /**
-     * @param \DateTimeInterface|null $startTime
-     *
-     * @return Job
-     */
     public function setStartTime(?\DateTimeInterface $startTime): Job
     {
         $this->startTime = $startTime;
@@ -361,19 +276,11 @@ class Job
         return $this;
     }
 
-    /**
-     * @return \DateTimeInterface|null
-     */
     public function getEndTime(): ?\DateTimeInterface
     {
         return $this->endTime;
     }
 
-    /**
-     * @param \DateTimeInterface|null $endTime
-     *
-     * @return Job
-     */
     public function setEndTime(?\DateTimeInterface $endTime): Job
     {
         $this->endTime = $endTime;

src/Entity/ScheduledDataflow.php

@@ -73,8 +73,7 @@ class ScheduledDataflow
     {
         $lost = array_diff(static::KEYS, array_keys($datas));
         if (count($lost) > 0) {
-            throw new \LogicException('The first argument of '.__METHOD__.'  must be contains: "'.implode(', ',
-                    $lost).'"');
+            throw new \LogicException('The first argument of '.__METHOD__.'  must be contains: "'.implode(', ', $lost).'"');
         }
 
         $scheduledDataflow = new self();
@@ -103,11 +102,6 @@ class ScheduledDataflow
         ];
     }
 
-    /**
-     * @param int $id
-     *
-     * @return ScheduledDataflow
-     */
     public function setId(int $id): ScheduledDataflow
     {
         $this->id = $id;
@@ -115,27 +109,16 @@ class ScheduledDataflow
         return $this;
     }
 
-    /**
-     * @return int|null
-     */
     public function getId(): ?int
     {
         return $this->id;
     }
 
-    /**
-     * @return string|null
-     */
     public function getLabel(): ?string
     {
         return $this->label;
     }
 
-    /**
-     * @param string|null $label
-     *
-     * @return ScheduledDataflow
-     */
     public function setLabel(?string $label): ScheduledDataflow
     {
         $this->label = $label;
@@ -143,19 +126,11 @@ class ScheduledDataflow
         return $this;
     }
 
-    /**
-     * @return string|null
-     */
     public function getDataflowType(): ?string
     {
         return $this->dataflowType;
     }
 
-    /**
-     * @param string|null $dataflowType
-     *
-     * @return ScheduledDataflow
-     */
     public function setDataflowType(?string $dataflowType): ScheduledDataflow
     {
         $this->dataflowType = $dataflowType;
@@ -163,19 +138,11 @@ class ScheduledDataflow
         return $this;
     }
 
-    /**
-     * @return array|null
-     */
     public function getOptions(): ?array
     {
         return $this->options;
     }
 
-    /**
-     * @param array|null $options
-     *
-     * @return ScheduledDataflow
-     */
     public function setOptions(?array $options): ScheduledDataflow
     {
         $this->options = $options;
@@ -183,19 +150,11 @@ class ScheduledDataflow
         return $this;
     }
 
-    /**
-     * @return string|null
-     */
     public function getFrequency(): ?string
     {
         return $this->frequency;
     }
 
-    /**
-     * @param string|null $frequency
-     *
-     * @return ScheduledDataflow
-     */
     public function setFrequency(?string $frequency): ScheduledDataflow
     {
         $this->frequency = $frequency;
@@ -203,19 +162,11 @@ class ScheduledDataflow
         return $this;
     }
 
-    /**
-     * @return \DateTimeInterface|null
-     */
     public function getNext(): ?\DateTimeInterface
     {
         return $this->next;
     }
 
-    /**
-     * @param \DateTimeInterface|null $next
-     *
-     * @return ScheduledDataflow
-     */
     public function setNext(?\DateTimeInterface $next): ScheduledDataflow
     {
         $this->next = $next;
@@ -223,19 +174,11 @@ class ScheduledDataflow
         return $this;
     }
 
-    /**
-     * @return bool|null
-     */
     public function getEnabled(): ?bool
     {
         return $this->enabled;
     }
 
-    /**
-     * @param bool|null $enabled
-     *
-     * @return ScheduledDataflow
-     */
     public function setEnabled(?bool $enabled): ScheduledDataflow
     {
         $this->enabled = $enabled;

src/Event/CrEvent.php

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\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
+    {
+    }
+}

src/Event/ProcessingEvent.php

@@ -5,31 +5,25 @@ declare(strict_types=1);
 namespace CodeRhapsodie\DataflowBundle\Event;
 
 use CodeRhapsodie\DataflowBundle\Entity\Job;
-use Symfony\Component\EventDispatcher\Event;
 
 /**
  * Event used during the dataflow lifecycle.
  *
  * @codeCoverageIgnore
  */
-class ProcessingEvent extends Event
+class ProcessingEvent extends CrEvent
 {
     /** @var Job */
     private $job;
 
     /**
      * ProcessingEvent constructor.
-     *
-     * @param Job $job
      */
     public function __construct(Job $job)
     {
         $this->job = $job;
     }
 
-    /**
-     * @return Job
-     */
     public function getJob(): Job
     {
         return $this->job;

src/Exceptions/UnknownDataflowTypeException.php

@@ -9,4 +9,12 @@ namespace CodeRhapsodie\DataflowBundle\Exceptions;
  */
 class UnknownDataflowTypeException extends \Exception
 {
+    public static function create(string $aliasOrFqcn, array $knownDataflowTypes): self
+    {
+        return new self(sprintf(
+            'Unknown dataflow type FQCN or alias "%s". Registered dataflow types FQCN and aliases are %s.',
+            $aliasOrFqcn,
+            implode(', ', $knownDataflowTypes)
+        ));
+    }
 }

src/Exceptions/UnsupportedItemTypeException.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\Exceptions;
+
+/**
+ * Exception thrown when a writer receives an item of an unsupported type.
+ */
+class UnsupportedItemTypeException extends \Exception
+{
+}

src/Logger/BufferHandler.php

@@ -0,0 +1,42 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\Logger;
+
+use Monolog\Formatter\FormatterInterface;
+use Monolog\Formatter\LineFormatter;
+use Monolog\Handler\AbstractProcessingHandler;
+use Monolog\Logger;
+
+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 = [];
+    }
+
+    public function clearBuffer(): array
+    {
+        $logs = $this->buffer;
+        $this->buffer = [];
+
+        return $logs;
+    }
+
+    protected function write(array $record): void
+    {
+        $this->buffer[] = $record['formatted'];
+    }
+
+    protected function getDefaultFormatter(): FormatterInterface
+    {
+        return new LineFormatter(self::FORMAT);
+    }
+}

src/Logger/DelegatingLogger.php

@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\Logger;
+
+use Psr\Log\AbstractLogger;
+use Psr\Log\LoggerInterface;
+
+final class DelegatingLogger extends AbstractLogger
+{
+    /** @var LoggerInterface[] */
+    private $loggers;
+
+    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)));
+            }
+
+            $this->loggers[] = $logger;
+        }
+    }
+
+    public function log($level, $message, array $context = [])
+    {
+        foreach ($this->loggers as $logger) {
+            $logger->log($level, $message, $context);
+        }
+    }
+}

src/Manager/ScheduledDataflowManager.php

@@ -4,10 +4,10 @@ declare(strict_types=1);
 
 namespace CodeRhapsodie\DataflowBundle\Manager;
 
-use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
 use CodeRhapsodie\DataflowBundle\Entity\Job;
-use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
+use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
 use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
+use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
 use Doctrine\DBAL\Driver\Connection;
 
 /**
@@ -53,9 +53,6 @@ class ScheduledDataflowManager implements ScheduledDataflowManagerInterface
         $this->connection->commit();
     }
 
-    /**
-     * @param ScheduledDataflow $scheduled
-     */
     private function updateScheduledDataflowNext(ScheduledDataflow $scheduled): void
     {
         $interval = \DateInterval::createFromDateString($scheduled->getFrequency());
@@ -67,11 +64,9 @@ class ScheduledDataflowManager implements ScheduledDataflowManagerInterface
         }
 
         $scheduled->setNext($next);
+        $this->scheduledDataflowRepository->save($scheduled);
     }
 
-    /**
-     * @param ScheduledDataflow $scheduled
-     */
     private function createPendingForScheduled(ScheduledDataflow $scheduled): void
     {
         $this->jobRepository->save(Job::createFromScheduledDataflow($scheduled));

src/Registry/DataflowTypeRegistry.php

@@ -4,8 +4,8 @@ declare(strict_types=1);
 
 namespace CodeRhapsodie\DataflowBundle\Registry;
 
-use CodeRhapsodie\DataflowBundle\Exceptions\UnknownDataflowTypeException;
 use CodeRhapsodie\DataflowBundle\DataflowType\DataflowTypeInterface;
+use CodeRhapsodie\DataflowBundle\Exceptions\UnknownDataflowTypeException;
 
 /**
  * Array based dataflow types registry.
@@ -31,7 +31,7 @@ class DataflowTypeRegistry implements DataflowTypeRegistryInterface
             return $this->aliasesRegistry[$fqcnOrAlias];
         }
 
-        throw new UnknownDataflowTypeException($fqcnOrAlias);
+        throw UnknownDataflowTypeException::create($fqcnOrAlias, array_merge(array_keys($this->fqcnRegistry), array_keys($this->aliasesRegistry)));
     }
 
     /**

src/Registry/DataflowTypeRegistryInterface.php

@@ -13,10 +13,6 @@ interface DataflowTypeRegistryInterface
 {
     /**
      * Get a registered dataflow type from its FQCN or one of its aliases.
-     *
-     * @param string $fqcnOrAlias
-     *
-     * @return DataflowTypeInterface
      */
     public function getDataflowType(string $fqcnOrAlias): DataflowTypeInterface;
 
@@ -29,8 +25,6 @@ interface DataflowTypeRegistryInterface
 
     /**
      * Registers a dataflow type.
-     *
-     * @param DataflowTypeInterface $dataflowType
      */
     public function registerDataflowType(DataflowTypeInterface $dataflowType): void;
 }

src/Repository/JobRepository.php

@@ -7,6 +7,7 @@ namespace CodeRhapsodie\DataflowBundle\Repository;
 use CodeRhapsodie\DataflowBundle\Entity\Job;
 use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
 use Doctrine\DBAL\Driver\Connection;
+use Doctrine\DBAL\ParameterType;
 use Doctrine\DBAL\Query\QueryBuilder;
 
 /**
@@ -21,15 +22,15 @@ class JobRepository
     public const TABLE_NAME = 'cr_dataflow_job';
 
     private const FIELDS_TYPE = [
-        'id' => \PDO::PARAM_INT,
-        'status' => \PDO::PARAM_INT,
-        'label' => \PDO::PARAM_STR,
-        'dataflow_type' => \PDO::PARAM_STR,
-        'options' => \PDO::PARAM_STR,
+        'id' => ParameterType::INTEGER,
+        'status' => ParameterType::INTEGER,
+        'label' => ParameterType::STRING,
+        'dataflow_type' => ParameterType::STRING,
+        'options' => ParameterType::STRING,
         'requested_date' => 'datetime',
-        'scheduled_dataflow_id' => \PDO::PARAM_INT,
-        'count' => \PDO::PARAM_INT,
-        'exceptions' => \PDO::PARAM_STR,
+        'scheduled_dataflow_id' => ParameterType::INTEGER,
+        'count' => ParameterType::INTEGER,
+        'exceptions' => ParameterType::STRING,
         'start_time' => 'datetime',
         'end_time' => 'datetime',
     ];
@@ -41,8 +42,6 @@ class JobRepository
 
     /**
      * JobRepository constructor.
-     *
-     * @param Connection $connection
      */
     public function __construct(Connection $connection)
     {
@@ -53,7 +52,7 @@ class JobRepository
     {
         $qb = $this->createQueryBuilder();
         $qb
-            ->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($jobId, \PDO::PARAM_INT)))
+            ->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($jobId, ParameterType::INTEGER)))
         ;
 
         return $this->returnFirstOrNull($qb);
@@ -64,12 +63,12 @@ class JobRepository
         $qb = $this->createQueryBuilder();
         $qb
             ->andWhere($qb->expr()->isNull('scheduled_dataflow_id'))
-            ->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, \PDO::PARAM_INT)));
+            ->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, ParameterType::INTEGER)));
         $stmt = $qb->execute();
         if (0 === $stmt->rowCount()) {
             return [];
         }
-        while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
+        while (false !== ($row = $stmt->fetchAssociative())) {
             yield Job::createFromArray($this->initDateTime($this->initArray($row)));
         }
     }
@@ -78,8 +77,8 @@ class JobRepository
     {
         $qb = $this->createQueryBuilder();
         $qb
-            ->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($scheduled->getId(), \PDO::PARAM_INT)))
-            ->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, \PDO::PARAM_INT)));
+            ->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($scheduled->getId(), ParameterType::INTEGER)))
+            ->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, ParameterType::INTEGER)));
 
         return $this->returnFirstOrNull($qb);
     }
@@ -88,7 +87,7 @@ class JobRepository
     {
         $qb = $this->createQueryBuilder();
         $qb->andWhere($qb->expr()->lte('requested_date', $qb->createNamedParameter(new \DateTime(), 'datetime')))
-            ->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, \PDO::PARAM_INT)))
+            ->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, ParameterType::INTEGER)))
             ->orderBy('requested_date', 'ASC')
             ->setMaxResults(1)
         ;
@@ -99,7 +98,7 @@ class JobRepository
     public function findLastForDataflowId(int $dataflowId): ?Job
     {
         $qb = $this->createQueryBuilder();
-        $qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($dataflowId, \PDO::PARAM_INT)))
+        $qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($dataflowId, ParameterType::INTEGER)))
             ->orderBy('requested_date', 'DESC')
             ->setMaxResults(1)
         ;
@@ -117,7 +116,7 @@ class JobRepository
         if (0 === $stmt->rowCount()) {
             return [];
         }
-        while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
+        while (false !== ($row = $stmt->fetchAssociative())) {
             yield Job::createFromArray($row);
         }
     }
@@ -125,14 +124,14 @@ class JobRepository
     public function findForScheduled(int $id): iterable
     {
         $qb = $this->createQueryBuilder();
-        $qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($id, \PDO::PARAM_INT)))
+        $qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($id, ParameterType::INTEGER)))
             ->orderBy('requested_date', 'DESC')
             ->setMaxResults(20);
         $stmt = $qb->execute();
         if (0 === $stmt->rowCount()) {
             return [];
         }
-        while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
+        while (false !== ($row = $stmt->fetchAssociative())) {
             yield Job::createFromArray($row);
         }
     }
@@ -174,6 +173,6 @@ class JobRepository
             return null;
         }
 
-        return Job::createFromArray($this->initDateTime($this->initArray($stmt->fetch(\PDO::FETCH_ASSOC))));
+        return Job::createFromArray($this->initDateTime($this->initArray($stmt->fetchAssociative())));
     }
 }

src/Repository/ScheduledDataflowRepository.php

@@ -6,6 +6,7 @@ namespace CodeRhapsodie\DataflowBundle\Repository;
 
 use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
 use Doctrine\DBAL\Driver\Connection;
+use Doctrine\DBAL\ParameterType;
 use Doctrine\DBAL\Query\QueryBuilder;
 
 /**
@@ -20,13 +21,13 @@ class ScheduledDataflowRepository
     public const TABLE_NAME = 'cr_dataflow_scheduled';
 
     private const FIELDS_TYPE = [
-        'id' => \PDO::PARAM_INT,
-        'label' => \PDO::PARAM_STR,
-        'dataflow_type' => \PDO::PARAM_STR,
-        'options' => \PDO::PARAM_STR,
-        'frequency' => \PDO::PARAM_STR,
+        'id' => ParameterType::INTEGER,
+        'label' => ParameterType::STRING,
+        'dataflow_type' => ParameterType::STRING,
+        'options' => ParameterType::STRING,
+        'frequency' => ParameterType::STRING,
         'next' => 'datetime',
-        'enabled' => \PDO::PARAM_BOOL,
+        'enabled' => ParameterType::BOOLEAN,
     ];
     /**
      * @var \Doctrine\DBAL\Connection
@@ -35,8 +36,6 @@ class ScheduledDataflowRepository
 
     /**
      * JobRepository constructor.
-     *
-     * @param Connection $connection
      */
     public function __construct(Connection $connection)
     {
@@ -60,7 +59,7 @@ class ScheduledDataflowRepository
         if (0 === $stmt->rowCount()) {
             return [];
         }
-        while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
+        while (false !== ($row = $stmt->fetchAssociative())) {
             yield ScheduledDataflow::createFromArray($this->initDateTime($this->initArray($row)));
         }
     }
@@ -68,7 +67,7 @@ class ScheduledDataflowRepository
     public function find(int $scheduleId): ?ScheduledDataflow
     {
         $qb = $this->createQueryBuilder();
-        $qb->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($scheduleId, \PDO::PARAM_INT)))
+        $qb->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($scheduleId, ParameterType::INTEGER)))
             ->setMaxResults(1)
         ;
 
@@ -84,7 +83,7 @@ class ScheduledDataflowRepository
         if (0 === $stmt->rowCount()) {
             return [];
         }
-        while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
+        while (false !== ($row = $stmt->fetchAssociative())) {
             yield ScheduledDataflow::createFromArray($this->initDateTime($this->initOptions($row)));
         }
     }
@@ -98,7 +97,7 @@ class ScheduledDataflowRepository
             ->orderBy('w.label', 'ASC')
             ->groupBy('w.id');
 
-        return $query->execute()->fetchAll(\PDO::FETCH_ASSOC);
+        return $query->execute()->fetchAllAssociative();
     }
 
     public function save(ScheduledDataflow $scheduledDataflow)
@@ -149,6 +148,6 @@ class ScheduledDataflowRepository
             return null;
         }
 
-        return ScheduledDataflow::createFromArray($this->initDateTime($this->initArray($stmt->fetch(\PDO::FETCH_ASSOC))));
+        return ScheduledDataflow::createFromArray($this->initDateTime($this->initArray($stmt->fetchAssociative())));
     }
 }

src/Resources/config/services.yaml

@@ -61,7 +61,7 @@ services:
 
     coderhapsodie.dataflow.connection.internal:
         lazy: true
-        class: Doctrine\DBAL\Driver\Connection
+        class: Doctrine\DBAL\Connection
         factory: ['@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory', 'getConnection']
 
     CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory:

src/Runner/PendingDataflowRunner.php

@@ -8,12 +8,20 @@ use CodeRhapsodie\DataflowBundle\DataflowType\Result;
 use CodeRhapsodie\DataflowBundle\Entity\Job;
 use CodeRhapsodie\DataflowBundle\Event\Events;
 use CodeRhapsodie\DataflowBundle\Event\ProcessingEvent;
+use CodeRhapsodie\DataflowBundle\Logger\BufferHandler;
+use CodeRhapsodie\DataflowBundle\Logger\DelegatingLogger;
 use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
 use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
+use Monolog\Logger;
+use Psr\Log\LoggerAwareInterface;
+use Psr\Log\LoggerAwareTrait;
+use Psr\Log\LoggerInterface;
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
-class PendingDataflowRunner implements PendingDataflowRunnerInterface
+class PendingDataflowRunner implements PendingDataflowRunnerInterface, LoggerAwareInterface
 {
+    use LoggerAwareTrait;
+
     /** @var JobRepository */
     private $repository;
 
@@ -39,18 +47,36 @@ class PendingDataflowRunner implements PendingDataflowRunnerInterface
             $this->beforeProcessing($job);
 
             $dataflowType = $this->registry->getDataflowType($job->getDataflowType());
+            $loggers = [new Logger('dataflow_internal', [$bufferHandler = new BufferHandler()])];
+            if (isset($this->logger)) {
+                $loggers[] = $this->logger;
+            }
+            $logger = new DelegatingLogger($loggers);
+
+            if ($dataflowType instanceof LoggerAwareInterface) {
+                $dataflowType->setLogger($logger);
+            }
+
             $result = $dataflowType->process($job->getOptions());
 
-            $this->afterProcessing($job, $result);
+            if (!$dataflowType instanceof LoggerAwareInterface) {
+                foreach ($result->getExceptions() as $index => $e) {
+                    $logger->error($e, ['index' => $index]);
+                }
+            }
+
+            $this->afterProcessing($job, $result, $bufferHandler);
         }
     }
 
-    /**
-     * @param Job $job
-     */
     private function beforeProcessing(Job $job): void
     {
-        $this->dispatcher->dispatch(Events::BEFORE_PROCESSING, new ProcessingEvent($job));
+        // 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);
+        }
 
         $job
             ->setStatus(Job::STATUS_RUNNING)
@@ -59,26 +85,21 @@ class PendingDataflowRunner implements PendingDataflowRunnerInterface
         $this->repository->save($job);
     }
 
-    /**
-     * @param Job    $job
-     * @param Result $result
-     */
-    private function afterProcessing(Job $job, Result $result): void
+    private function afterProcessing(Job $job, Result $result, BufferHandler $bufferLogger): void
     {
-        $exceptions = [];
-        /** @var \Exception $exception */
-        foreach ($result->getExceptions() as $exception) {
-            $exceptions[] = (string) $exception;
-        }
-
         $job
             ->setEndTime($result->getEndTime())
             ->setStatus(Job::STATUS_COMPLETED)
             ->setCount($result->getSuccessCount())
-            ->setExceptions($exceptions)
+            ->setExceptions($bufferLogger->clearBuffer())
         ;
         $this->repository->save($job);
 
-        $this->dispatcher->dispatch(Events::AFTER_PROCESSING, new ProcessingEvent($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);
+        }
     }
 }

src/SchemaProvider/DataflowSchemaProvider.php

@@ -19,10 +19,10 @@ class DataflowSchemaProvider
     {
         $schema = new Schema();
         $tableJob = $schema->createTable(JobRepository::TABLE_NAME);
-        $tableJob->addColumn('id', 'integer', array(
+        $tableJob->addColumn('id', 'integer', [
             'autoincrement' => true,
-        ));
-        $tableJob->setPrimaryKey(array('id'));
+        ]);
+        $tableJob->setPrimaryKey(['id']);
 
         $tableJob->addColumn('scheduled_dataflow_id', 'integer', ['notnull' => false]);
         $tableJob->addColumn('status', 'integer', ['notnull' => true]);
@@ -36,10 +36,10 @@ class DataflowSchemaProvider
         $tableJob->addColumn('end_time', 'datetime', ['notnull' => false]);
 
         $tableSchedule = $schema->createTable(ScheduledDataflowRepository::TABLE_NAME);
-        $tableSchedule->addColumn('id', 'integer', array(
+        $tableSchedule->addColumn('id', 'integer', [
             'autoincrement' => true,
-        ));
-        $tableSchedule->setPrimaryKey(array('id'));
+        ]);
+        $tableSchedule->setPrimaryKey(['id']);
         $tableSchedule->addColumn('label', 'string', ['notnull' => true, 'length' => 255]);
         $tableSchedule->addColumn('dataflow_type', 'string', ['notnull' => true, 'length' => 255]);
         $tableSchedule->addColumn('options', 'json', ['notnull' => true]);