add depentency to doctrine/doctrine-bundle (#29)

.travis.yml

@@ -23,91 +23,67 @@ 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: '8.0'
+    - php: '7.3'
       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.3'
+    - php: '7.1'
       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'
+    - php: '7.1'
       env:
         - SYMFONY_VERSION=4.2.*
-    - php: '7.3'
+    - php: '7.1'
       env:
         - SYMFONY_VERSION=4.3.*
-    - php: '7.3'
+
+    # Test unsupported versions of Symfony
+    - php: '7.1'
       env:
-        - SYMFONY_VERSION=5.0.*
-    - php: '7.3'
+        - SYMFONY_VERSION=4.0.*
+    - php: '7.1'
       env:
-        - SYMFONY_VERSION=5.1.*
+        - SYMFONY_VERSION=4.1.*
+
 
     # Test upcoming Symfony versions with lowest supported PHP version and dev dependencies
-#    - php: '7.2'
-#      env:
-#        - STABILITY=dev
-#        - SYMFONY_VERSION=5.3.*
+    - php: '7.1'
+      env:
+        - STABILITY=dev
+        - SYMFONY_VERSION=4.4.*
 
     # Test upcoming PHP versions with dev dependencies
-    #- php: '7.5snapshot'
-    #  env:
-    #    - STABILITY=dev
-    #    - COMPOSER_FLAGS="--ignore-platform-reqs --prefer-stable"
+    - php: '7.4snapshot'
+      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.*
+        - SYMFONY_VERSION=4.4.*
 
 before_install:
   - if [[ "$SYMFONY_VERSION" != "" ]]; then
@@ -121,15 +97,15 @@ before_install:
       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;
+      travis_retry composer require --dev satooshi/php-coveralls:^2.0 --no-update $COMPOSER_FLAGS;
     fi
 
 install:
-  - travis_retry composer update --prefer-dist --no-interaction --no-progress --ansi $COMPOSER_FLAGS
+  - travis_retry composer update --prefer-dist --no-interaction --no-suggest --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,53 +0,0 @@
-# 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
-* Add configuration to define the default Doctrine DBAL connection
-* Remove Doctrine ORM
-* Rewrite repositories
-
-# Version 1.0.1
-
-* Fix lost dependency
-* Fix schedule removing
-
-# Version 1.0.0
-
-Initial version
-
-* Define and configure a Dataflow
-* Run the Job scheduled
-* Run one Dataflow from the command line
-* Define the schedule for a Dataflow from the command line
-* Enable/Disable a scheduled Dataflow from the command line
-* Display the list of scheduled Dataflow from the command line
-* Display the result for the last Job for a Dataflow from the command line

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.com/code-rhapsodie/dataflow-bundle.svg?branch=master)](https://travis-ci.com/code-rhapsodie/dataflow-bundle)
+[![Build Status](https://travis-ci.org/code-rhapsodie/dataflow-bundle.svg?branch=master)](https://travis-ci.org/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)
 
@@ -30,21 +30,16 @@ As the following schema shows, you can define more than one dataflow:
 * Enable/Disable a scheduled Dataflow from the command line
 * Display the list of scheduled Dataflow from the command line
 * 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 :
 
 ```shell script
-$ composer require code-rhapsodie/dataflow-bundle
+$ composer require code-rhapsodie/dataflow
 ```
 
 #### Suggest
@@ -103,40 +98,26 @@ public function registerBundles()
 
 ### Update the database
 
-This bundle uses Doctrine DBAL to store Dataflow schedule into the database table (`cr_dataflow_scheduled`)
+This bundle uses Doctrine ORM for drive the database table for store Dataflow schedule (`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/) 
-or [Kaliop Migration Bundle](https://github.com/kaliop-uk/ezmigrationbundle) or whatever,
+#### Doctrine migration
+
+Execute the command to generate the migration for your database:
+
+```shell script
+$ bin/console doctrine:migration:diff
+```
+
+#### Other migration tools
+
+If you use [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:
 
 ```shell script
-$ bin/console code-rhapsodie:dataflow:dump-schema
+$ bin/console doctrine:schema:update --dump-sql
 ```
 
-If you have already the tables, you can add a new migration with the generated update SQL query from this command:
-
-```shell script
-$ bin/console code-rhapsodie:dataflow:dump-schema --update
-```
-
-## Configuration
-
-By default, the Doctrine DBAL connection used is `default`. You can configure the default connection.
-Add this configuration into your Symfony configuration:
-
-```yaml
-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
 
@@ -179,11 +160,11 @@ class MyFirstDataflowType extends AbstractDataflowType
 
     protected function buildDataflow(DataflowBuilder $builder, array $options): void
     {
-        $this->myWriter->setDestinationFilePath($options['to-file']);
+        
+        $this->myReader->setFilename($options['fileName']);
 
-        $builder
-            ->setReader($this->myReader->read($options['from-file']))
-            ->addStep(function ($data) use ($options) {
+        $builder->setReader($this->myReader)
+            ->addStep(function($data) use ($options) {
                 // TODO : Write your code here...
                 return $data;
             })
@@ -193,8 +174,11 @@ class MyFirstDataflowType extends AbstractDataflowType
 
     protected function configureOptions(OptionsResolver $optionsResolver): void
     {
-        $optionsResolver->setDefaults(['to-file' => '/tmp/dataflow.csv', 'from-file' => null]);
-        $optionsResolver->setRequired('from-file');
+        $optionsResolver->setDefaults([
+            'my_option' => 'my_default_value',
+            'fileName'  => null,
+        ]);
+        $optionsResolver->setRequired('fileName');
     }
 
     public function getLabel(): string
@@ -238,8 +222,11 @@ class MyFirstDataflowType extends AbstractDataflowType
     // ...
     protected function configureOptions(OptionsResolver $optionsResolver): void
     {
-        $optionsResolver->setDefaults(['to-file' => '/tmp/dataflow.csv', 'from-file' => null]);
-        $optionsResolver->setRequired('from-file');
+        $optionsResolver->setDefaults([
+            'my_option' => 'my_default_value',
+            'fileName'  => null,
+        ]);
+        $optionsResolver->setRequired('fileName');
     }
 
 }
@@ -247,30 +234,6 @@ 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:
@@ -311,18 +274,27 @@ namespace CodeRhapsodie\DataflowExemple\Reader;
 
 class FileReader
 {
-    public function read(string $filename): iterable
+    private $filename;
+
+    /**
+     * Set the filename option needed by the Reader.
+     */
+    public function setFilename(string $filename) {
+        $this->filename = $filename;
+    }
+
+    public function __invoke(): iterable
     {
-        if (!$filename) {
+        if (!$this->filename) {
             throw new \Exception("The file name is not defined. Define it with 'setFilename' method");
         }
 
-        if (!$fh = fopen($filename, 'r')) {
-            throw new \Exception("Unable to open file '".$filename."' for read.");
+        if (!$fh = fopen($this->filename, 'r')) {
+            throw new \Exception("Unable to open file '".$this->filename."' for read.");
         }
 
-        while (false !== ($read = fgets($fh))) {
-            yield explode('|', trim($read));
+        while (false === ($read = fread($fh, 1024))) {
+            yield explode("|", $read);
         }
     }
 }
@@ -348,16 +320,14 @@ A *Step* can be any callable, taking the element as its argument, and returning
 A few examples:
 
 ```php
-<?php
-//[...]
-$builder->addStep(function ($item) {
+$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;
@@ -365,7 +335,6 @@ $builder->addStep(function ($item) {
 
     return $item;
 });
-//[...]
 ```
 
 ### Writers
@@ -393,20 +362,11 @@ class FileWriter implements WriterInterface
 {
     private $fh;
 
-    /** @var string */
-    private $path;
-
-    public function setDestinationFilePath(string $path) {
-        $this->path = $path;
-    }
-
     public function prepare()
     {
-        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.');
+
+        if (!$this->fh = fopen('/path/to/file', 'w')) {
+            throw new \Exception("Unable to open in write mode the output file.");
         }
     }
 
@@ -422,95 +382,6 @@ 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.
@@ -537,28 +408,6 @@ Several commands are provided to manage schedules and run jobs.
 
 `code-rhapsodie:dataflow:execute` Let you execute one dataflow job.
 
-`code-rhapsodie:dataflow:dump-schema` Generates schema create / update SQL queries
-
-### Work with many databases
-
-All commands have a `--connection` option to define what Doctrine DBAL connection to use during execution.
-
-Example:
-
-This command uses the `default` DBAL connection to generate all schema update queries.
-
-```shell script
-$ bin/console code-rhapsodie:dataflow:dump-schema --update --connection=default
-```
-
-To execute all pending job for a specific connection use:
-
-```shell script
-# Run for dataflow DBAL connection
-$ bin/console code-rhapsodie:dataflow:run-pending --connection=dataflow
-# Run for default DBAL connection
-$ bin/console code-rhapsodie:dataflow:run-pending --connection=default
-```
 
 # Issues and feature requests
 

Tests/DataflowType/AbstractDataflowTypeTest.php

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

Tests/DataflowType/Writer/CollectionWriterTest.php

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

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

@@ -9,7 +9,6 @@ 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 PHPUnit\Framework\MockObject\MockObject;
 use PHPUnit\Framework\TestCase;
@@ -19,8 +18,8 @@ class ScheduledDataflowManagerTest extends TestCase
     /** @var ScheduledDataflowManager */
     private $manager;
 
-    /** @var Connection|MockObject */
-    private $connection;
+    /** @var EntityManagerInterface|MockObject */
+    private $em;
 
     /** @var ScheduledDataflowRepository|MockObject */
     private $scheduledDataflowRepository;
@@ -30,18 +29,17 @@ class ScheduledDataflowManagerTest extends TestCase
 
     protected function setUp(): void
     {
-        $this->connection = $this->createMock(Connection::class);
+        $this->em = $this->createMock(EntityManagerInterface::class);
         $this->scheduledDataflowRepository = $this->createMock(ScheduledDataflowRepository::class);
         $this->jobRepository = $this->createMock(JobRepository::class);
 
-        $this->manager = new ScheduledDataflowManager($this->connection, $this->scheduledDataflowRepository, $this->jobRepository);
+        $this->manager = new ScheduledDataflowManager($this->em, $this->scheduledDataflowRepository, $this->jobRepository);
     }
 
     public function testCreateJobsFromScheduledDataflows()
     {
         $scheduled1 = new ScheduledDataflow();
         $scheduled2 = (new ScheduledDataflow())
-            ->setId(-1)
             ->setDataflowType($type = 'testType')
             ->setOptions($options = ['opt' => 'val'])
             ->setNext($next = new \DateTime())
@@ -62,13 +60,9 @@ class ScheduledDataflowManagerTest extends TestCase
             ->willReturnOnConsecutiveCalls(new Job(), null)
         ;
 
-        $this->connection
+        $this->em
             ->expects($this->once())
-            ->method('beginTransaction')
-        ;
-        $this->jobRepository
-            ->expects($this->once())
-            ->method('save')
+            ->method('persist')
             ->with(
                 $this->callback(function (Job $job) use ($type, $options, $next, $label, $scheduled2) {
                     return (
@@ -77,65 +71,19 @@ class ScheduledDataflowManagerTest extends TestCase
                         && $job->getOptions() === $options
                         && $job->getRequestedDate() == $next
                         && $job->getLabel() === $label
-                        && $job->getScheduledDataflowId() === $scheduled2->getId()
+                        && $job->getScheduledDataflow() === $scheduled2
                     );
                 })
             )
         ;
 
-        $this->scheduledDataflowRepository
+        $this->em
             ->expects($this->once())
-            ->method('save')
-            ->with($scheduled2)
-        ;
-
-        $this->connection
-            ->expects($this->once())
-            ->method('commit')
+            ->method('flush')
         ;
 
         $this->manager->createJobsFromScheduledDataflows();
 
         $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

@@ -20,6 +20,9 @@ class PendingDataflowRunnerTest extends TestCase
     /** @var PendingDataflowRunner */
     private $runner;
 
+    /** @var EntityManagerInterface|MockObject */
+    private $em;
+
     /** @var JobRepository|MockObject */
     private $repository;
 
@@ -31,11 +34,12 @@ class PendingDataflowRunnerTest extends TestCase
 
     protected function setUp(): void
     {
+        $this->em = $this->createMock(EntityManagerInterface::class);
         $this->repository = $this->createMock(JobRepository::class);
         $this->registry = $this->createMock(DataflowTypeRegistryInterface::class);
         $this->dispatcher = $this->createMock(EventDispatcherInterface::class);
 
-        $this->runner = new PendingDataflowRunner($this->repository, $this->registry, $this->dispatcher);
+        $this->runner = new PendingDataflowRunner($this->em, $this->repository, $this->registry, $this->dispatcher);
     }
 
     public function testRunPendingDataflows()
@@ -58,68 +62,36 @@ class PendingDataflowRunnerTest extends TestCase
             ->willReturnOnConsecutiveCalls($job1, $job2, null)
         ;
 
-        // 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,
-                    ]
-                );
-        }
+        $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;
+                    })
+                ]
+            )
+        ;
 
         $dataflowType1 = $this->createMock(DataflowTypeInterface::class);
         $dataflowType2 = $this->createMock(DataflowTypeInterface::class);
@@ -150,9 +122,9 @@ class PendingDataflowRunnerTest extends TestCase
             ->willReturn($result2)
         ;
 
-        $this->repository
+        $this->em
             ->expects($this->exactly(4))
-            ->method('save')
+            ->method('flush')
         ;
 
         $this->runner->runPendingDataflows();

UPGRADE.md

@@ -1,3 +0,0 @@
-# Upgrade from v1.x to v2.0
-
-[BC] `JobRepository` and `ScheduledDataflowRepository` are no longer a Doctrine ORM repository.

composer.json

@@ -2,12 +2,7 @@
     "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": [
         {
@@ -41,23 +36,22 @@
         }
     },
     "require": {
-        "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"
+        "php": "^7.1",
+        "doctrine/orm": "^2.4.5",
+        "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"
     },
     "require-dev": {
+        "friendsofphp/php-cs-fixer": "^2.15",
         "phpunit/phpunit": "^7||^8"
     },
     "suggest": {
@@ -68,9 +62,7 @@
     },
     "extra": {
         "branch-alias": {
-            "dev-master": "2.1.x-dev",
-            "dev-v2.0.x": "2.0.x-dev",
-            "dev-v1.x": "1.x-dev"
+            "dev-master": "1.x-dev"
         }
     }
 }

src/CodeRhapsodieDataflowBundle.php

@@ -6,7 +6,6 @@ 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;
 
@@ -24,9 +23,6 @@ class CodeRhapsodieDataflowBundle extends Bundle
 
     public function build(ContainerBuilder $container)
     {
-        $container
-            ->addCompilerPass(new DataflowTypeCompilerPass())
-            ->addCompilerPass(new DefaultLoggerCompilerPass())
-        ;
+        $container->addCompilerPass(new DataflowTypeCompilerPass());
     }
 }

src/Command/AddScheduledDataflowCommand.php

@@ -5,7 +5,6 @@ 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;
@@ -29,17 +28,13 @@ class AddScheduledDataflowCommand extends Command
     /** @var ValidatorInterface */
     private $validator;
 
-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(DataflowTypeRegistryInterface $registry, ScheduledDataflowRepository $scheduledDataflowRepository, ValidatorInterface $validator, ConnectionFactory $connectionFactory)
+    public function __construct(DataflowTypeRegistryInterface $registry, ScheduledDataflowRepository $scheduledDataflowRepository, ValidatorInterface $validator)
     {
         parent::__construct();
 
         $this->registry = $registry;
         $this->scheduledDataflowRepository = $scheduledDataflowRepository;
         $this->validator = $validator;
-        $this->connectionFactory = $connectionFactory;
     }
 
     /**
@@ -52,12 +47,10 @@ class AddScheduledDataflowCommand extends Command
             ->setHelp('The <info>%command.name%</info> allows you to create a new scheduled dataflow.')
             ->addOption('label', null, InputOption::VALUE_REQUIRED, 'Label of the scheduled dataflow')
             ->addOption('type', null, InputOption::VALUE_REQUIRED, 'Type of the scheduled dataflow (FQCN)')
-            ->addOption('options', null, InputOption::VALUE_OPTIONAL,
-                'Options of the scheduled dataflow (ex: {"option1": "value1", "option2": "value2"})')
+            ->addOption('options', null, InputOption::VALUE_OPTIONAL, 'Options of the scheduled dataflow (ex: {"option1": "value1", "option2": "value2"})')
             ->addOption('frequency', null, InputOption::VALUE_REQUIRED, 'Frequency of the scheduled dataflow')
             ->addOption('first_run', null, InputOption::VALUE_REQUIRED, 'Date for the first run of the scheduled dataflow (Y-m-d H:i:s)')
-            ->addOption('enabled', null, InputOption::VALUE_REQUIRED, 'State of the scheduled dataflow')
-            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
+            ->addOption('enabled', null, InputOption::VALUE_REQUIRED, 'State of the scheduled dataflow');
     }
 
     /**
@@ -65,9 +58,6 @@ class AddScheduledDataflowCommand extends Command
      */
     protected function execute(InputInterface $input, OutputInterface $output)
     {
-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
         $choices = [];
         $typeMapping = [];
         foreach ($this->registry->listDataflowTypes() as $fqcn => $dataflowType) {
@@ -87,13 +77,11 @@ class AddScheduledDataflowCommand extends Command
         }
         $options = $input->getOption('options');
         if (!$options) {
-            $options = $io->ask('What are the launch options for the scheduled dataflow? (ex: {"option1": "value1", "option2": "value2"})',
-                json_encode([]));
+            $options = $io->ask('What are the launch options for the scheduled dataflow? (ex: {"option1": "value1", "option2": "value2"})', json_encode([]));
         }
         $frequency = $input->getOption('frequency');
         if (!$frequency) {
-            $frequency = $io->choice('What is the frequency for the scheduled dataflow?',
-                ScheduledDataflow::AVAILABLE_FREQUENCIES);
+            $frequency = $io->choice('What is the frequency for the scheduled dataflow?', ScheduledDataflow::AVAILABLE_FREQUENCIES);
         }
         $firstRun = $input->getOption('first_run');
         if (!$firstRun) {
@@ -104,27 +92,44 @@ class AddScheduledDataflowCommand extends Command
             $enabled = $io->confirm('Enable the scheduled dataflow?');
         }
 
-        $newScheduledDataflow = ScheduledDataflow::createFromArray([
-            'id' => null,
-            'label' => $label,
-            'dataflow_type' => $type,
-            'options' => json_decode($options, true),
-            'frequency' => $frequency,
-            'next' => new \DateTimeImmutable($firstRun),
-            'enabled' => $enabled,
-        ]);
+        try {
+            $newScheduledDataflow = $this->createEntityFromArray([
+                'label' => $label,
+                'type' => $type,
+                'options' => $options,
+                'frequency' => $frequency,
+                'first_run' => $firstRun,
+                'enabled' => $enabled,
+            ]);
 
-        $errors = $this->validator->validate($newScheduledDataflow);
-        if (count($errors) > 0) {
-            $io->error((string) $errors);
+            $errors = $this->validator->validate($newScheduledDataflow);
+            if (count($errors) > 0) {
+                $io->error((string) $errors);
 
-            return 2;
+                return 2;
+            }
+
+            $this->scheduledDataflowRepository->save($newScheduledDataflow);
+            $io->success(sprintf('New scheduled dataflow "%s" (id:%d) was created successfully.', $newScheduledDataflow->getLabel(), $newScheduledDataflow->getId()));
+
+            return 0;
+        } catch (\Exception $e) {
+            $io->error(sprintf('An error occured when creating new scheduled dataflow : "%s".', $e->getMessage()));
+
+            return 1;
         }
+    }
 
-        $this->scheduledDataflowRepository->save($newScheduledDataflow);
-        $io->success(sprintf('New scheduled dataflow "%s" (id:%d) was created successfully.',
-            $newScheduledDataflow->getLabel(), $newScheduledDataflow->getId()));
+    private function createEntityFromArray(array $input): ScheduledDataflow
+    {
+        $scheduledDataflow = new ScheduledDataflow();
+        $scheduledDataflow->setLabel($input['label']);
+        $scheduledDataflow->setDataflowType($input['type']);
+        $scheduledDataflow->setOptions(json_decode($input['options'], true));
+        $scheduledDataflow->setFrequency($input['frequency']);
+        $scheduledDataflow->setNext(new \DateTimeImmutable($input['first_run']));
+        $scheduledDataflow->setEnabled($input['enabled']);
 
-        return 0;
+        return $scheduledDataflow;
     }
 }

src/Command/ChangeScheduleStatusCommand.php

@@ -5,7 +5,6 @@ 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;
@@ -24,15 +23,11 @@ class ChangeScheduleStatusCommand extends Command
     /** @var ScheduledDataflowRepository */
     private $scheduledDataflowRepository;
 
-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(ScheduledDataflowRepository $scheduledDataflowRepository, ConnectionFactory $connectionFactory)
+    public function __construct(ScheduledDataflowRepository $scheduledDataflowRepository)
     {
         parent::__construct();
 
         $this->scheduledDataflowRepository = $scheduledDataflowRepository;
-        $this->connectionFactory = $connectionFactory;
     }
 
     /**
@@ -45,8 +40,7 @@ class ChangeScheduleStatusCommand extends Command
             ->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')
-            ->addOption('disable', null, InputOption::VALUE_NONE, 'Disable the schedule')
-            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
+            ->addOption('disable', null, InputOption::VALUE_NONE, 'Disable the schedule');
     }
 
     /**
@@ -54,12 +48,9 @@ class ChangeScheduleStatusCommand extends Command
      */
     protected function execute(InputInterface $input, OutputInterface $output)
     {
-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
         $io = new SymfonyStyle($input, $output);
         /** @var ScheduledDataflow|null $schedule */
-        $schedule = $this->scheduledDataflowRepository->find((int) $input->getArgument('schedule-id'));
+        $schedule = $this->scheduledDataflowRepository->find($input->getArgument('schedule-id'));
 
         if (!$schedule) {
             $io->error(sprintf('Cannot find scheduled dataflow with id "%d".', $input->getArgument('schedule-id')));

src/Command/ExecuteDataflowCommand.php

@@ -4,41 +4,29 @@ 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 Symfony\Component\Console\Style\SymfonyStyle;
 
 /**
  * Runs one dataflow.
  *
  * @codeCoverageIgnore
  */
-class ExecuteDataflowCommand extends Command implements LoggerAwareInterface
+class ExecuteDataflowCommand extends Command
 {
-    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(DataflowTypeRegistryInterface $registry)
     {
         parent::__construct();
 
         $this->registry = $registry;
-        $this->connectionFactory = $connectionFactory;
     }
 
     /**
@@ -56,7 +44,7 @@ EOF
             )
             ->addArgument('fqcn', InputArgument::REQUIRED, 'FQCN or alias of the dataflow type')
             ->addArgument('options', InputArgument::OPTIONAL, 'Options for the dataflow type as a json string', '[]')
-            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
+        ;
     }
 
     /**
@@ -64,30 +52,16 @@ EOF
      */
     protected function execute(InputInterface $input, OutputInterface $output)
     {
-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
         $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);
 
-        $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;
-        }
+        $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());
 
         return 0;
     }

src/Command/JobShowCommand.php

@@ -5,7 +5,6 @@ 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;
@@ -29,15 +28,11 @@ class JobShowCommand extends Command
     /** @var JobRepository */
     private $jobRepository;
 
-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(JobRepository $jobRepository, ConnectionFactory $connectionFactory)
+    public function __construct(JobRepository $jobRepository)
     {
         parent::__construct();
 
         $this->jobRepository = $jobRepository;
-        $this->connectionFactory = $connectionFactory;
     }
 
     /**
@@ -50,8 +45,7 @@ class JobShowCommand extends Command
             ->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')
-            ->addOption('details', null, InputOption::VALUE_NONE, 'Display full details')
-            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
+            ->addOption('details', null, InputOption::VALUE_NONE, 'Display full details');
     }
 
     /**
@@ -59,10 +53,6 @@ class JobShowCommand extends Command
      */
     protected function execute(InputInterface $input, OutputInterface $output)
     {
-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
-
         $io = new SymfonyStyle($input, $output);
 
         $jobId = (int) $input->getOption('job-id');

src/Command/RunPendingDataflowsCommand.php

@@ -4,13 +4,11 @@ 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;
 use Symfony\Component\Console\Command\LockableTrait;
 use Symfony\Component\Console\Input\InputInterface;
-use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 
 /**
@@ -30,16 +28,12 @@ class RunPendingDataflowsCommand extends Command
     /** @var PendingDataflowRunnerInterface */
     private $runner;
 
-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(ScheduledDataflowManagerInterface $manager, PendingDataflowRunnerInterface $runner, ConnectionFactory $connectionFactory)
+    public function __construct(ScheduledDataflowManagerInterface $manager, PendingDataflowRunnerInterface $runner)
     {
         parent::__construct();
 
         $this->manager = $manager;
         $this->runner = $runner;
-        $this->connectionFactory = $connectionFactory;
     }
 
     /**
@@ -53,7 +47,7 @@ class RunPendingDataflowsCommand extends Command
 The <info>%command.name%</info> command runs dataflows according to the schedule defined in the UI by the user.
 EOF
             )
-            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
+        ;
     }
 
     /**
@@ -67,10 +61,6 @@ EOF
             return 0;
         }
 
-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
-
         $this->manager->createJobsFromScheduledDataflows();
         $this->runner->runPendingDataflows();
 

src/Command/ScheduleListCommand.php

@@ -4,11 +4,9 @@ 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;
 
@@ -22,15 +20,11 @@ class ScheduleListCommand extends Command
     /** @var ScheduledDataflowRepository */
     private $scheduledDataflowRepository;
 
-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(ScheduledDataflowRepository $scheduledDataflowRepository, ConnectionFactory $connectionFactory)
+    public function __construct(ScheduledDataflowRepository $scheduledDataflowRepository)
     {
         parent::__construct();
 
         $this->scheduledDataflowRepository = $scheduledDataflowRepository;
-        $this->connectionFactory = $connectionFactory;
     }
 
     /**
@@ -40,8 +34,7 @@ class ScheduleListCommand extends Command
     {
         $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');
+            ->setHelp('The <info>%command.name%</info> lists all scheduled dataflows.');
     }
 
     /**
@@ -49,9 +42,6 @@ class ScheduleListCommand extends Command
      */
     protected function execute(InputInterface $input, OutputInterface $output)
     {
-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
         $io = new SymfonyStyle($input, $output);
         $display = [];
         $schedules = $this->scheduledDataflowRepository->listAllOrderedByLabel();
@@ -61,7 +51,7 @@ class ScheduleListCommand extends Command
                 $schedule['label'],
                 $schedule['enabled'] ? 'yes' : 'no',
                 $schedule['startTime'] ? (new \DateTime($schedule['startTime']))->format('Y-m-d H:i:s') : '-',
-                $schedule['next'] ? (new \DateTime($schedule['next']))->format('Y-m-d H:i:s') : '-',
+                $schedule['next'] ? $schedule['next']->format('Y-m-d H:i:s') : '-',
             ];
         }
 

src/Command/SchemaCommand.php

@@ -1,101 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace CodeRhapsodie\DataflowBundle\Command;
-
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
-use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
-use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
-use CodeRhapsodie\DataflowBundle\SchemaProvider\DataflowSchemaProvider;
-use Doctrine\DBAL\Schema\Schema;
-use Doctrine\DBAL\Schema\Table;
-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;
-
-/**
- * @codeCoverageIgnore
- */
-class SchemaCommand extends Command
-{
-    protected static $defaultName = 'code-rhapsodie:dataflow:dump-schema';
-
-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(ConnectionFactory $connectionFactory)
-    {
-        parent::__construct();
-
-        $this->connectionFactory = $connectionFactory;
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    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')
-        ;
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    protected function execute(InputInterface $input, OutputInterface $output)
-    {
-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
-
-        $connection = $this->connectionFactory->getConnection();
-
-        $schemaProvider = new DataflowSchemaProvider();
-        $schema = $schemaProvider->createSchema();
-
-        $sqls = $schema->toSql($connection->getDatabasePlatform());
-
-        if ($input->getOption('update')) {
-            $sm = $connection->getSchemaManager();
-
-            $tableArray = [JobRepository::TABLE_NAME, ScheduledDataflowRepository::TABLE_NAME];
-            $tables = [];
-            foreach ($sm->listTables() as $table) {
-                /** @var Table $table */
-                if (in_array($table->getName(), $tableArray)) {
-                    $tables[] = $table;
-                }
-            }
-
-            $namespaces = [];
-
-            if ($connection->getDatabasePlatform()->supportsSchemas()) {
-                $namespaces = $sm->listNamespaceNames();
-            }
-
-            $sequences = [];
-
-            if ($connection->getDatabasePlatform()->supportsSequences()) {
-                $sequences = $sm->listSequences();
-            }
-
-            $oldSchema = new Schema($tables, $sequences, $sm->createSchemaConfig(), $namespaces);
-
-            $sqls = $schema->getMigrateFromSql($oldSchema, $connection->getDatabasePlatform());
-        }
-        $io = new SymfonyStyle($input, $output);
-        $io->text('Execute these SQL Queries on your database:');
-        foreach ($sqls as $sql) {
-            $io->text($sql.';');
-        }
-
-        return 0;
-    }
-}

src/DataflowType/AbstractDataflowType.php

@@ -4,15 +4,10 @@ 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, LoggerAwareInterface
+abstract class AbstractDataflowType implements DataflowTypeInterface
 {
-    use LoggerAwareTrait;
-
     /**
      * @codeCoverageIgnore
      */
@@ -32,9 +27,6 @@ abstract class AbstractDataflowType implements DataflowTypeInterface, LoggerAwar
         ;
         $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,13 +6,10 @@ namespace CodeRhapsodie\DataflowBundle\DataflowType\Dataflow;
 
 use CodeRhapsodie\DataflowBundle\DataflowType\Result;
 use CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface;
-use Psr\Log\LoggerAwareInterface;
-use Psr\Log\LoggerAwareTrait;
+use CodeRhapsodie\DataflowBundle\Exceptions\InterruptedProcessingException;
 
-class Dataflow implements DataflowInterface, LoggerAwareInterface
+class Dataflow implements DataflowInterface
 {
-    use LoggerAwareTrait;
-
     /** @var string */
     private $name;
 
@@ -20,20 +17,24 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface
     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,6 +45,8 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface
     }
 
     /**
+     * @param WriterInterface $writer
+     *
      * @return $this
      */
     public function addWriter(WriterInterface $writer): self
@@ -60,33 +63,27 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface
     {
         $count = 0;
         $exceptions = [];
-        $startTime = new \DateTimeImmutable();
+        $startTime = new \DateTime();
 
-        try {
-            foreach ($this->writers as $writer) {
-                $writer->prepare();
-            }
-
-            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->prepare();
         }
 
-        return new Result($this->name, $startTime, new \DateTimeImmutable(), $count, $exceptions);
+        foreach ($this->reader as $index => $item) {
+            try {
+                $this->processItem($item);
+            } catch (\Exception $e) {
+                $exceptions[$index] = $e;
+            }
+
+            ++$count;
+        }
+
+        foreach ($this->writers as $writer) {
+            $writer->finish();
+        }
+
+        return new Result($this->name, $startTime, new \DateTime(), $count, $exceptions);
     }
 
     /**
@@ -106,13 +103,4 @@ class Dataflow implements DataflowInterface, LoggerAwareInterface
             $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,6 +13,8 @@ interface DataflowInterface
 {
     /**
      * Processes the data.
+     *
+     * @return Result
      */
     public function process(): Result;
 }

src/DataflowType/Result.php

@@ -33,6 +33,13 @@ 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;
@@ -45,46 +52,73 @@ 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

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

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

@@ -1,96 +0,0 @@
-<?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,25 +4,11 @@ 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

@@ -24,10 +24,5 @@ class CodeRhapsodieDataflowExtension extends Extension
             ->registerForAutoconfiguration(DataflowTypeInterface::class)
             ->addTag('coderhapsodie.dataflow.type')
         ;
-        $configuration = new Configuration();
-        $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

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

@@ -1,35 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace CodeRhapsodie\DataflowBundle\DependencyInjection;
-
-use Symfony\Component\Config\Definition\Builder\TreeBuilder;
-use Symfony\Component\Config\Definition\ConfigurationInterface;
-
-class Configuration implements ConfigurationInterface
-{
-    public function getConfigTreeBuilder()
-    {
-        $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()
-        ;
-
-        return $treeBuilder;
-    }
-}

src/Entity/Job.php

@@ -4,11 +4,14 @@ declare(strict_types=1);
 
 namespace CodeRhapsodie\DataflowBundle\Entity;
 
-use Symfony\Component\Validator\Constraints as Asserts;
+use Doctrine\ORM\Mapping as ORM;
 
 /**
  * Dataflow execution status.
  *
+ * @ORM\Entity(repositoryClass="CodeRhapsodie\DataflowBundle\Repository\JobRepository")
+ * @ORM\Table(name="cr_dataflow_job")
+ *
  * @codeCoverageIgnore
  */
 class Job
@@ -17,86 +20,89 @@ class Job
     const STATUS_RUNNING = 1;
     const STATUS_COMPLETED = 2;
 
-    private const KEYS = [
-        'id',
-        'status',
-        'label',
-        'dataflow_type',
-        'options',
-        'requested_date',
-        'scheduled_dataflow_id',
-        'count',
-        'exceptions',
-        'start_time',
-        'end_time',
-    ];
-
     /**
-     * @var int|null
+     * @var int
+     *
+     * @ORM\Id()
+     * @ORM\Column(type="integer")
+     * @ORM\GeneratedValue(strategy="AUTO")
      */
     private $id;
 
     /**
      * @var int
      *
-     * @Asserts\Range(min=0, max=2)
+     * @ORM\Column(type="integer")
      */
     private $status;
 
     /**
      * @var string|null
      *
-     * @Asserts\NotBlank()
-     * @Asserts\Length(min=1, max=255)
-     * @Asserts\Regex("#^[[:alnum:] ]+\z#u")
+     * @ORM\Column(type="string")
      */
     private $label;
 
     /**
      * @var string|null
      *
-     * @Asserts\NotBlank()
-     * @Asserts\Length(min=1, max=255)
-     * @Asserts\Regex("#^[[:alnum:]\\]+\z#u")
+     * @ORM\Column(type="string")
      */
     private $dataflowType;
 
     /**
      * @var array|null
+     *
+     * @ORM\Column(type="json")
      */
     private $options;
 
     /**
      * @var \DateTimeInterface|null
+     *
+     * @ORM\Column(type="datetime", nullable=true)
      */
     private $requestedDate;
 
     /**
-     * @var int|null
+     * @var ScheduledDataflow|null
+     *
+     * @ORM\ManyToOne(targetEntity="ScheduledDataflow", inversedBy="jobs")
+     * @ORM\JoinColumn(nullable=true)
      */
-    private $scheduledDataflowId;
+    private $scheduledDataflow;
 
     /**
      * @var int|null
+     *
+     * @ORM\Column(type="integer", nullable=true)
      */
     private $count;
 
     /**
      * @var array|null
+     *
+     * @ORM\Column(type="json", nullable=true)
      */
     private $exceptions;
 
     /**
      * @var \DateTimeInterface|null
+     *
+     * @ORM\Column(type="datetime", nullable=true)
      */
     private $startTime;
 
     /**
      * @var \DateTimeInterface|null
+     *
+     * @ORM\Column(type="datetime", nullable=true)
      */
     private $endTime;
 
     /**
+     * @param ScheduledDataflow $scheduled
+     *
      * @return Job
      */
     public static function createFromScheduledDataflow(ScheduledDataflow $scheduled): self
@@ -107,72 +113,31 @@ class Job
             ->setOptions($scheduled->getOptions())
             ->setRequestedDate(clone $scheduled->getNext())
             ->setLabel($scheduled->getLabel())
-            ->setScheduledDataflowId($scheduled->getId());
+            ->setScheduledDataflow($scheduled)
+        ;
     }
 
-    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));
-        if (count($lost) > 0) {
-            throw new \LogicException('The first argument of '.__METHOD__.'  must be contains: "'.implode(', ', $lost).'"');
-        }
-
-        $job = new self();
-        $job->id = null === $datas['id'] ? null : (int) $datas['id'];
-        $job->setStatus(null === $datas['status'] ? null : (int) $datas['status']);
-        $job->setLabel($datas['label']);
-        $job->setDataflowType($datas['dataflow_type']);
-        $job->setOptions($datas['options']);
-        $job->setRequestedDate($datas['requested_date']);
-        $job->setScheduledDataflowId(null === $datas['scheduled_dataflow_id'] ? null : (int) $datas['scheduled_dataflow_id']);
-        $job->setCount(null === $datas['count'] ? null : (int) $datas['count']);
-        $job->setExceptions($datas['exceptions']);
-        $job->setStartTime($datas['start_time']);
-        $job->setEndTime($datas['end_time']);
-
-        return $job;
-    }
-
-    public function toArray(): array
-    {
-        return [
-            'id' => $this->getId(),
-            'status' => $this->getStatus(),
-            'label' => $this->getLabel(),
-            'dataflow_type' => $this->getDataflowType(),
-            'options' => $this->getOptions(),
-            'requested_date' => $this->getRequestedDate(),
-            'scheduled_dataflow_id' => $this->getScheduledDataflowId(),
-            'count' => $this->getCount(),
-            'exceptions' => $this->getExceptions(),
-            'start_time' => $this->getStartTime(),
-            'end_time' => $this->getEndTime(),
-        ];
-    }
-
-    public function setId(int $id): Job
-    {
-        $this->id = $id;
-
-        return $this;
-    }
-
-    public function getId(): ?int
+    /**
+     * @return int
+     */
+    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;
@@ -180,11 +145,19 @@ 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;
@@ -192,11 +165,19 @@ 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;
@@ -204,11 +185,19 @@ 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;
@@ -216,11 +205,19 @@ 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;
@@ -228,23 +225,39 @@ class Job
         return $this;
     }
 
-    public function getScheduledDataflowId(): ?int
+    /**
+     * @return ScheduledDataflow|null
+     */
+    public function getScheduledDataflow(): ?ScheduledDataflow
     {
-        return $this->scheduledDataflowId;
+        return $this->scheduledDataflow;
     }
 
-    public function setScheduledDataflowId(?int $scheduledDataflowId): Job
+    /**
+     * @param ScheduledDataflow|null $scheduledDataflow
+     *
+     * @return Job
+     */
+    public function setScheduledDataflow(?ScheduledDataflow $scheduledDataflow): Job
     {
-        $this->scheduledDataflowId = $scheduledDataflowId;
+        $this->scheduledDataflow = $scheduledDataflow;
 
         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;
@@ -252,11 +265,19 @@ 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;
@@ -264,11 +285,19 @@ 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;
@@ -276,11 +305,19 @@ 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

@@ -5,11 +5,14 @@ declare(strict_types=1);
 namespace CodeRhapsodie\DataflowBundle\Entity;
 
 use CodeRhapsodie\DataflowBundle\Validator\Constraints\Frequency;
-use Symfony\Component\Validator\Constraints as Asserts;
+use Doctrine\ORM\Mapping as ORM;
 
 /**
  * Schedule for a regular execution of a dataflow.
  *
+ * @ORM\Entity(repositoryClass="CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository")
+ * @ORM\Table(name="cr_dataflow_scheduled")
+ *
  * @codeCoverageIgnore
  */
 class ScheduledDataflow
@@ -21,104 +24,88 @@ class ScheduledDataflow
         '1 month',
     ];
 
-    private const KEYS = ['id', 'label', 'dataflow_type', 'options', 'frequency', 'next', 'enabled'];
-
     /**
-     * @var int|null
+     * @var int
+     *
+     * @ORM\Id()
+     * @ORM\Column(name="id", type="integer")
+     * @ORM\GeneratedValue(strategy="AUTO")
      */
     private $id;
 
     /**
      * @var string|null
      *
-     * @Asserts\NotBlank()
-     * @Asserts\Length(min=1, max=255)
-     * @Asserts\Regex("#^[[:alnum:] ]+\z#u")
+     * @ORM\Column(type="string")
      */
     private $label;
 
     /**
      * @var string|null
      *
-     * @Asserts\NotBlank()
-     * @Asserts\Length(min=1, max=255)
-     * @Asserts\Regex("#^[[:alnum:]\\]+\z#u")
+     * @ORM\Column(type="string")
      */
     private $dataflowType;
 
     /**
      * @var array|null
+     *
+     * @ORM\Column(type="json")
      */
     private $options;
 
     /**
      * @var string|null
      *
-     * @Asserts\NotBlank()
+     * @ORM\Column(type="string")
+     *
      * @Frequency()
      */
     private $frequency;
 
     /**
      * @var \DateTimeInterface|null
+     *
+     * @ORM\Column(type="datetime", nullable=true)
      */
     private $next;
 
     /**
      * @var bool|null
+     *
+     * @ORM\Column(type="boolean")
      */
     private $enabled;
 
-    public static function createFromArray(array $datas)
-    {
-        $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).'"');
-        }
+    /**
+     * @var Job[]
+     *
+     * @ORM\OneToMany(targetEntity="Job", mappedBy="scheduledDataflow", cascade={"persist", "remove"})
+     * @ORM\OrderBy({"startTime" = "DESC"})
+     */
+    private $jobs;
 
-        $scheduledDataflow = new self();
-        $scheduledDataflow->id = null === $datas['id'] ? null : (int) $datas['id'];
-
-        $scheduledDataflow->setLabel($datas['label']);
-        $scheduledDataflow->setDataflowType($datas['dataflow_type']);
-        $scheduledDataflow->setOptions($datas['options']);
-        $scheduledDataflow->setFrequency($datas['frequency']);
-        $scheduledDataflow->setNext($datas['next']);
-        $scheduledDataflow->setEnabled(null === $datas['enabled'] ? null : (bool) $datas['enabled']);
-
-        return $scheduledDataflow;
-    }
-
-    public function toArray(): array
-    {
-        return [
-            'id' => $this->getId(),
-            'label' => $this->getLabel(),
-            'dataflow_type' => $this->getDataflowType(),
-            'options' => $this->getOptions(),
-            'frequency' => $this->getFrequency(),
-            'next' => $this->getNext(),
-            'enabled' => $this->getEnabled(),
-        ];
-    }
-
-    public function setId(int $id): ScheduledDataflow
-    {
-        $this->id = $id;
-
-        return $this;
-    }
-
-    public function getId(): ?int
+    /**
+     * @return int
+     */
+    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;
@@ -126,11 +113,19 @@ 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;
@@ -138,11 +133,19 @@ 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;
@@ -150,11 +153,19 @@ 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;
@@ -162,11 +173,19 @@ 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;
@@ -174,11 +193,19 @@ 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

@@ -1,20 +0,0 @@
-<?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,25 +5,31 @@ 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 CrEvent
+class ProcessingEvent extends Event
 {
     /** @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,12 +9,4 @@ 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

@@ -1,12 +0,0 @@
-<?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/Factory/ConnectionFactory.php

@@ -1,35 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace CodeRhapsodie\DataflowBundle\Factory;
-
-use Symfony\Component\DependencyInjection\Container;
-
-/**
- * Class ConnectionFactory.
- *
- * @codeCoverageIgnore
- */
-class ConnectionFactory
-{
-    private $connectionName;
-
-    private $container;
-
-    public function __construct(Container $container, string $connectionName)
-    {
-        $this->connectionName = $connectionName;
-        $this->container = $container;
-    }
-
-    public function setConnectionName(string $connectionName)
-    {
-        $this->connectionName = $connectionName;
-    }
-
-    public function getConnection(): \Doctrine\DBAL\Driver\Connection
-    {
-        return $this->container->get(sprintf('doctrine.dbal.%s_connection', $this->connectionName));
-    }
-}

src/Logger/BufferHandler.php

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

@@ -1,32 +0,0 @@
-<?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,29 +4,29 @@ declare(strict_types=1);
 
 namespace CodeRhapsodie\DataflowBundle\Manager;
 
-use CodeRhapsodie\DataflowBundle\Entity\Job;
 use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
-use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
+use CodeRhapsodie\DataflowBundle\Entity\Job;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
-use Doctrine\DBAL\Driver\Connection;
+use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
+use Doctrine\ORM\EntityManagerInterface;
 
 /**
  * Handles scheduled dataflows execution dates based on their frequency.
  */
 class ScheduledDataflowManager implements ScheduledDataflowManagerInterface
 {
+    /** @var EntityManagerInterface */
+    private $em;
+
     /** @var ScheduledDataflowRepository */
     private $scheduledDataflowRepository;
 
     /** @var JobRepository */
     private $jobRepository;
 
-    /** @var Connection */
-    private $connection;
-
-    public function __construct(Connection $connection, ScheduledDataflowRepository $scheduledDataflowRepository, JobRepository $jobRepository)
+    public function __construct(EntityManagerInterface $em, ScheduledDataflowRepository $scheduledDataflowRepository, JobRepository $jobRepository)
     {
-        $this->connection = $connection;
+        $this->em = $em;
         $this->scheduledDataflowRepository = $scheduledDataflowRepository;
         $this->jobRepository = $jobRepository;
     }
@@ -36,23 +36,21 @@ class ScheduledDataflowManager implements ScheduledDataflowManagerInterface
      */
     public function createJobsFromScheduledDataflows(): void
     {
-        $this->connection->beginTransaction();
-        try {
-            foreach ($this->scheduledDataflowRepository->findReadyToRun() as $scheduled) {
-                if (null !== $this->jobRepository->findPendingForScheduledDataflow($scheduled)) {
-                    continue;
-                }
-
-                $this->createPendingForScheduled($scheduled);
-                $this->updateScheduledDataflowNext($scheduled);
+        foreach ($this->scheduledDataflowRepository->findReadyToRun() as $scheduled) {
+            if (null !== $this->jobRepository->findPendingForScheduledDataflow($scheduled)) {
+                continue;
             }
-        } catch (\Throwable $e) {
-            $this->connection->rollBack();
-            throw $e;
+
+            $this->createPendingForScheduled($scheduled);
+            $this->updateScheduledDataflowNext($scheduled);
         }
-        $this->connection->commit();
+
+        $this->em->flush();
     }
 
+    /**
+     * @param ScheduledDataflow $scheduled
+     */
     private function updateScheduledDataflowNext(ScheduledDataflow $scheduled): void
     {
         $interval = \DateInterval::createFromDateString($scheduled->getFrequency());
@@ -64,11 +62,13 @@ 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));
+        $this->em->persist(Job::createFromScheduledDataflow($scheduled));
     }
 }

src/Registry/DataflowTypeRegistry.php

@@ -4,8 +4,8 @@ declare(strict_types=1);
 
 namespace CodeRhapsodie\DataflowBundle\Registry;
 
-use CodeRhapsodie\DataflowBundle\DataflowType\DataflowTypeInterface;
 use CodeRhapsodie\DataflowBundle\Exceptions\UnknownDataflowTypeException;
+use CodeRhapsodie\DataflowBundle\DataflowType\DataflowTypeInterface;
 
 /**
  * Array based dataflow types registry.
@@ -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 new UnknownDataflowTypeException();
     }
 
     /**

src/Registry/DataflowTypeRegistryInterface.php

@@ -13,6 +13,10 @@ 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;
 
@@ -25,6 +29,8 @@ interface DataflowTypeRegistryInterface
 
     /**
      * Registers a dataflow type.
+     *
+     * @param DataflowTypeInterface $dataflowType
      */
     public function registerDataflowType(DataflowTypeInterface $dataflowType): void;
 }

src/Repository/InitFromDbTrait.php

@@ -1,45 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace CodeRhapsodie\DataflowBundle\Repository;
-
-/**
- * @codeCoverageIgnore
- */
-trait InitFromDbTrait
-{
-    private function initDateTime(array $datas): array
-    {
-        foreach (static::FIELDS_TYPE as $key => $type) {
-            if ('datetime' === $type && null !== $datas[$key]) {
-                $datas[$key] = new \DateTime($datas[$key]);
-            }
-        }
-
-        return $datas;
-    }
-
-    private function initArray(array $datas): array
-    {
-        if (!is_array($datas['options'])) {
-            $datas['options'] = $this->strToArray($datas['options']);
-        }
-        if (array_key_exists('exceptions', $datas) && !is_array($datas['exceptions'])) {
-            $datas['exceptions'] = $this->strToArray($datas['exceptions']);
-        }
-
-        return $datas;
-    }
-
-    private function strToArray($value): array
-    {
-        if (null === $value) {
-            return [];
-        }
-
-        $array = json_decode($value, true);
-
-        return (false === $array) ? [] : $array;
-    }
-}

src/Repository/JobRepository.php

@@ -4,175 +4,64 @@ declare(strict_types=1);
 
 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;
+use CodeRhapsodie\DataflowBundle\Entity\Job;
+use Doctrine\Common\Collections\Criteria;
+use Doctrine\ORM\EntityRepository;
 
 /**
  * Repository.
  *
  * @codeCoverageIgnore
  */
-class JobRepository
+class JobRepository extends EntityRepository
 {
-    use InitFromDbTrait;
-
-    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)
-    {
-        $this->connection = $connection;
-    }
-
-    public function find(int $jobId)
-    {
-        $qb = $this->createQueryBuilder();
-        $qb
-            ->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($jobId, ParameterType::INTEGER)))
-        ;
-
-        return $this->returnFirstOrNull($qb);
-    }
-
     public function findOneshotDataflows(): iterable
     {
-        $qb = $this->createQueryBuilder();
-        $qb
-            ->andWhere($qb->expr()->isNull('scheduled_dataflow_id'))
-            ->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, ParameterType::INTEGER)));
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
-            return [];
-        }
-        while (false !== ($row = $stmt->fetchAssociative())) {
-            yield Job::createFromArray($this->initDateTime($this->initArray($row)));
-        }
+        return $this->findBy([
+            'scheduledDataflow' => null,
+            'status' => Job::STATUS_PENDING,
+        ]);
     }
 
     public function findPendingForScheduledDataflow(ScheduledDataflow $scheduled): ?Job
     {
-        $qb = $this->createQueryBuilder();
-        $qb
-            ->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);
+        return $this->findOneBy([
+            'scheduledDataflow' => $scheduled->getId(),
+            'status' => Job::STATUS_PENDING,
+        ]);
     }
 
     public function findNextPendingDataflow(): ?Job
     {
-        $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, ParameterType::INTEGER)))
-            ->orderBy('requested_date', 'ASC')
+        $criteria = (new Criteria())
+            ->where(Criteria::expr()->lte('requestedDate', new \DateTime()))
+            ->andWhere(Criteria::expr()->eq('status', Job::STATUS_PENDING))
+            ->orderBy(['requestedDate' => Criteria::ASC])
             ->setMaxResults(1)
         ;
 
-        return $this->returnFirstOrNull($qb);
+        return $this->matching($criteria)->first() ?: null;
     }
 
     public function findLastForDataflowId(int $dataflowId): ?Job
     {
-        $qb = $this->createQueryBuilder();
-        $qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($dataflowId, ParameterType::INTEGER)))
-            ->orderBy('requested_date', 'DESC')
-            ->setMaxResults(1)
-        ;
-
-        return $this->returnFirstOrNull($qb);
+        return $this->findOneBy(['scheduledDataflow' => $dataflowId], ['requestedDate' => 'desc']);
     }
 
     public function findLatests(): iterable
     {
-        $qb = $this->createQueryBuilder();
-        $qb
-            ->orderBy('requested_date', 'DESC')
-            ->setMaxResults(20);
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
-            return [];
-        }
-        while (false !== ($row = $stmt->fetchAssociative())) {
-            yield Job::createFromArray($row);
-        }
+        return $this->findBy([], ['requestedDate' => 'desc'], 20);
     }
 
     public function findForScheduled(int $id): iterable
     {
-        $qb = $this->createQueryBuilder();
-        $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->fetchAssociative())) {
-            yield Job::createFromArray($row);
-        }
+        return $this->findBy(['scheduledDataflow' => $id], ['requestedDate' => 'desc'], 20);
     }
 
     public function save(Job $job)
     {
-        $datas = $job->toArray();
-        unset($datas['id']);
-
-        if (is_array($datas['options'])) {
-            $datas['options'] = json_encode($datas['options']);
-        }
-        if (is_array($datas['exceptions'])) {
-            $datas['exceptions'] = json_encode($datas['exceptions']);
-        }
-
-        if (null === $job->getId()) {
-            $this->connection->insert(static::TABLE_NAME, $datas, static::FIELDS_TYPE);
-            $job->setId((int) $this->connection->lastInsertId());
-
-            return;
-        }
-        $this->connection->update(static::TABLE_NAME, $datas, ['id' => $job->getId()], static::FIELDS_TYPE);
-    }
-
-    public function createQueryBuilder($alias = null): QueryBuilder
-    {
-        $qb = $this->connection->createQueryBuilder();
-        $qb->select('*')
-            ->from(static::TABLE_NAME, $alias);
-
-        return $qb;
-    }
-
-    private function returnFirstOrNull(QueryBuilder $qb): ?Job
-    {
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
-            return null;
-        }
-
-        return Job::createFromArray($this->initDateTime($this->initArray($stmt->fetchAssociative())));
+        $this->_em->persist($job);
+        $this->_em->flush();
     }
 }

src/Repository/ScheduledDataflowRepository.php

@@ -5,43 +5,16 @@ declare(strict_types=1);
 namespace CodeRhapsodie\DataflowBundle\Repository;
 
 use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
-use Doctrine\DBAL\Driver\Connection;
-use Doctrine\DBAL\ParameterType;
-use Doctrine\DBAL\Query\QueryBuilder;
+use Doctrine\Common\Collections\Criteria;
+use Doctrine\ORM\EntityRepository;
 
 /**
  * Repository for the ScheduledDataflow entity.
  *
  * @codeCoverageIgnore
  */
-class ScheduledDataflowRepository
+class ScheduledDataflowRepository extends EntityRepository
 {
-    use InitFromDbTrait;
-
-    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)
-    {
-        $this->connection = $connection;
-    }
-
     /**
      * Finds all enabled scheduled dataflows with a passed next run date.
      *
@@ -49,105 +22,42 @@ class ScheduledDataflowRepository
      */
     public function findReadyToRun(): iterable
     {
-        $qb = $this->createQueryBuilder();
-        $qb->andWhere($qb->expr()->lte('next', $qb->createNamedParameter(new \DateTime(), 'datetime')))
-            ->andWhere($qb->expr()->eq('enabled', 1))
-            ->orderBy('next', 'ASC')
-            ;
-
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
-            return [];
-        }
-        while (false !== ($row = $stmt->fetchAssociative())) {
-            yield ScheduledDataflow::createFromArray($this->initDateTime($this->initArray($row)));
-        }
-    }
-
-    public function find(int $scheduleId): ?ScheduledDataflow
-    {
-        $qb = $this->createQueryBuilder();
-        $qb->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($scheduleId, ParameterType::INTEGER)))
-            ->setMaxResults(1)
+        $criteria = (new Criteria())
+            ->where(Criteria::expr()->lte('next', new \DateTime()))
+            ->andWhere(Criteria::expr()->eq('enabled', 1))
+            ->orderBy(['next' => Criteria::ASC])
         ;
 
-        return $this->returnFirstOrNull($qb);
+        return $this->matching($criteria);
     }
 
     public function findAllOrderedByLabel(): iterable
     {
-        $qb = $this->createQueryBuilder();
-        $qb->orderBy('label', 'ASC');
-
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
-            return [];
-        }
-        while (false !== ($row = $stmt->fetchAssociative())) {
-            yield ScheduledDataflow::createFromArray($this->initDateTime($this->initOptions($row)));
-        }
+        return $this->findBy([], ['label' => 'asc']);
     }
 
     public function listAllOrderedByLabel(): array
     {
-        $query = $this->connection->createQueryBuilder()
-            ->from(static::TABLE_NAME, 'w')
-            ->select('w.id', 'w.label', 'w.enabled', 'w.next', 'max(j.start_time) as startTime')
-            ->leftJoin('w', JobRepository::TABLE_NAME, 'j', 'j.scheduled_dataflow_id = w.id')
+        $query = $this->createQueryBuilder('w')
+            ->select('w.id', 'w.label', 'w.enabled', 'w.next', 'max(j.startTime) as startTime')
+            ->leftJoin('w.jobs', 'j')
             ->orderBy('w.label', 'ASC')
             ->groupBy('w.id');
 
-        return $query->execute()->fetchAllAssociative();
+        return $query->getQuery()->execute();
     }
 
     public function save(ScheduledDataflow $scheduledDataflow)
     {
-        $datas = $scheduledDataflow->toArray();
-        unset($datas['id']);
-
-        if (is_array($datas['options'])) {
-            $datas['options'] = json_encode($datas['options']);
-        }
-
-        if (null === $scheduledDataflow->getId()) {
-            $this->connection->insert(static::TABLE_NAME, $datas, static::FIELDS_TYPE);
-            $scheduledDataflow->setId((int) $this->connection->lastInsertId());
-
-            return;
-        }
-        $this->connection->update(static::TABLE_NAME, $datas, ['id' => $scheduledDataflow->getId()], static::FIELDS_TYPE);
+        $this->_em->persist($scheduledDataflow);
+        $this->_em->flush();
     }
 
     public function delete(int $id): void
     {
-        $this->connection->beginTransaction();
-        try {
-            $this->connection->delete(JobRepository::TABLE_NAME, ['scheduled_dataflow_id' => $id]);
-            $this->connection->delete(static::TABLE_NAME, ['id' => $id]);
-        } catch (\Throwable $e) {
-            $this->connection->rollBack();
-            throw $e;
-        }
+        $dataflow = $this->find($id);
 
-        $this->connection->commit();
-    }
-
-    public function createQueryBuilder($alias = null): QueryBuilder
-    {
-        $qb = $this->connection->createQueryBuilder();
-        $qb->select('*')
-            ->from(static::TABLE_NAME, $alias);
-
-        return $qb;
-    }
-
-    private function returnFirstOrNull(QueryBuilder $qb): ?ScheduledDataflow
-    {
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
-            return null;
-        }
-
-        return ScheduledDataflow::createFromArray($this->initDateTime($this->initArray($stmt->fetchAssociative())));
+        $this->_em->remove($dataflow);
+        $this->_em->flush();
     }
 }

src/Resources/config/services.yaml

@@ -10,73 +10,53 @@ services:
             $registry: '@CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface'
             $scheduledDataflowRepository: '@CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository'
             $validator: '@validator'
-            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
         tags: ['console.command']
 
     CodeRhapsodie\DataflowBundle\Command\ChangeScheduleStatusCommand:
         arguments:
             $scheduledDataflowRepository: '@CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository'
-            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
         tags: ['console.command']
 
     CodeRhapsodie\DataflowBundle\Command\ExecuteDataflowCommand:
         arguments:
             $registry: '@CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface'
-            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
         tags: ['console.command']
 
     CodeRhapsodie\DataflowBundle\Command\JobShowCommand:
         arguments:
             $jobRepository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
-            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
         tags: ['console.command']
 
     CodeRhapsodie\DataflowBundle\Command\RunPendingDataflowsCommand:
         arguments:
             $manager: '@CodeRhapsodie\DataflowBundle\Manager\ScheduledDataflowManagerInterface'
             $runner: '@CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunnerInterface'
-            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
         tags: ['console.command']
 
     CodeRhapsodie\DataflowBundle\Command\ScheduleListCommand:
         arguments:
             $scheduledDataflowRepository: '@CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository'
-            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
-        tags: ['console.command']
-
-    CodeRhapsodie\DataflowBundle\Command\SchemaCommand:
-        arguments:
-            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
         tags: ['console.command']
 
     CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository:
-        lazy: true
-        arguments: ['@coderhapsodie.dataflow.connection']
+        factory: ['@doctrine.orm.default_entity_manager', 'getRepository']
+        arguments: ['CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow']
 
     CodeRhapsodie\DataflowBundle\Repository\JobRepository:
-        lazy: true
-        arguments: ['@coderhapsodie.dataflow.connection']
-
-    coderhapsodie.dataflow.connection: "@coderhapsodie.dataflow.connection.internal"
-
-    coderhapsodie.dataflow.connection.internal:
-        lazy: true
-        class: Doctrine\DBAL\Connection
-        factory: ['@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory', 'getConnection']
-
-    CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory:
-        arguments: ['@service_container', '%coderhapsodie.dataflow.dbal_default_connection%']
+        factory: ['@doctrine.orm.default_entity_manager', 'getRepository']
+        arguments: ['CodeRhapsodie\DataflowBundle\Entity\Job']
 
     CodeRhapsodie\DataflowBundle\Manager\ScheduledDataflowManagerInterface: '@CodeRhapsodie\DataflowBundle\Manager\ScheduledDataflowManager'
     CodeRhapsodie\DataflowBundle\Manager\ScheduledDataflowManager:
         arguments:
-            $connection: '@coderhapsodie.dataflow.connection'
+            $em: '@doctrine.orm.default_entity_manager'
             $scheduledDataflowRepository: '@CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository'
             $jobRepository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
 
     CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunnerInterface: '@CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunner'
     CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunner:
         arguments:
+            $em: '@doctrine.orm.default_entity_manager'
             $repository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
             $registry: '@CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface'
             $dispatcher: '@event_dispatcher'

src/Runner/PendingDataflowRunner.php

@@ -8,19 +8,15 @@ 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 Doctrine\ORM\EntityManagerInterface;
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
-class PendingDataflowRunner implements PendingDataflowRunnerInterface, LoggerAwareInterface
+class PendingDataflowRunner implements PendingDataflowRunnerInterface
 {
-    use LoggerAwareTrait;
+    /** @var EntityManagerInterface */
+    private $em;
 
     /** @var JobRepository */
     private $repository;
@@ -31,8 +27,9 @@ class PendingDataflowRunner implements PendingDataflowRunnerInterface, LoggerAwa
     /** @var EventDispatcherInterface */
     private $dispatcher;
 
-    public function __construct(JobRepository $repository, DataflowTypeRegistryInterface $registry, EventDispatcherInterface $dispatcher)
+    public function __construct(EntityManagerInterface $em, JobRepository $repository, DataflowTypeRegistryInterface $registry, EventDispatcherInterface $dispatcher)
     {
+        $this->em = $em;
         $this->repository = $repository;
         $this->registry = $registry;
         $this->dispatcher = $dispatcher;
@@ -47,59 +44,46 @@ class PendingDataflowRunner implements PendingDataflowRunnerInterface, LoggerAwa
             $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());
 
-            if (!$dataflowType instanceof LoggerAwareInterface) {
-                foreach ($result->getExceptions() as $index => $e) {
-                    $logger->error($e, ['index' => $index]);
-                }
-            }
-
-            $this->afterProcessing($job, $result, $bufferHandler);
+            $this->afterProcessing($job, $result);
         }
     }
 
+    /**
+     * @param Job $job
+     */
     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(Events::BEFORE_PROCESSING, new ProcessingEvent($job));
 
         $job
             ->setStatus(Job::STATUS_RUNNING)
             ->setStartTime(new \DateTime())
         ;
-        $this->repository->save($job);
+        $this->em->flush();
     }
 
-    private function afterProcessing(Job $job, Result $result, BufferHandler $bufferLogger): void
+    /**
+     * @param Job    $job
+     * @param Result $result
+     */
+    private function afterProcessing(Job $job, Result $result): 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($bufferLogger->clearBuffer())
+            ->setExceptions($exceptions)
         ;
-        $this->repository->save($job);
+        $this->em->flush();
 
-        // 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(Events::AFTER_PROCESSING, new ProcessingEvent($job));
     }
 }

src/SchemaProvider/DataflowSchemaProvider.php

@@ -1,54 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace CodeRhapsodie\DataflowBundle\SchemaProvider;
-
-use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
-use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
-use Doctrine\DBAL\Schema\Schema;
-
-/**
- * Class JobSchemaProvider.
- *
- * @codeCoverageIgnore
- */
-class DataflowSchemaProvider
-{
-    public function createSchema()
-    {
-        $schema = new Schema();
-        $tableJob = $schema->createTable(JobRepository::TABLE_NAME);
-        $tableJob->addColumn('id', 'integer', [
-            'autoincrement' => true,
-        ]);
-        $tableJob->setPrimaryKey(['id']);
-
-        $tableJob->addColumn('scheduled_dataflow_id', 'integer', ['notnull' => false]);
-        $tableJob->addColumn('status', 'integer', ['notnull' => true]);
-        $tableJob->addColumn('label', 'string', ['notnull' => true, 'length' => 255]);
-        $tableJob->addColumn('dataflow_type', 'string', ['notnull' => true, 'length' => 255]);
-        $tableJob->addColumn('options', 'json', ['notnull' => true]);
-        $tableJob->addColumn('requested_date', 'datetime', ['notnull' => false]);
-        $tableJob->addColumn('count', 'integer', ['notnull' => false]);
-        $tableJob->addColumn('exceptions', 'json', ['notnull' => false]);
-        $tableJob->addColumn('start_time', 'datetime', ['notnull' => false]);
-        $tableJob->addColumn('end_time', 'datetime', ['notnull' => false]);
-
-        $tableSchedule = $schema->createTable(ScheduledDataflowRepository::TABLE_NAME);
-        $tableSchedule->addColumn('id', 'integer', [
-            'autoincrement' => true,
-        ]);
-        $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]);
-        $tableSchedule->addColumn('frequency', 'string', ['notnull' => true, 'length' => 255]);
-        $tableSchedule->addColumn('next', 'datetime', ['notnull' => false]);
-        $tableSchedule->addColumn('enabled', 'boolean', ['notnull' => true]);
-
-        $tableJob->addForeignKeyConstraint($tableSchedule, ['scheduled_dataflow_id'], ['id']);
-
-        return $schema;
-    }
-}