mirror of
https://github.com/code-rhapsodie/dataflow-bundle.git
synced 2026-03-24 06:42:23 +01:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
4c80b3c51b |
43
.travis.yml
43
.travis.yml
@@ -26,19 +26,10 @@ matrix:
|
||||
- php: '7.1'
|
||||
- php: '7.2'
|
||||
- php: '7.3'
|
||||
- php: '7.4'
|
||||
|
||||
# Enable code coverage with the previous supported PHP version
|
||||
- php: '7.3'
|
||||
env:
|
||||
- SYMFONY_VERSION=3.4.*
|
||||
- COVERALLS_ENABLED="true"
|
||||
- PHPUNIT_FLAGS="-v --coverage-text --coverage-clover var/build/clover.xml"
|
||||
|
||||
# Enable code coverage with the latest supported PHP version
|
||||
- php: '7.4'
|
||||
- php: '7.3'
|
||||
env:
|
||||
- SYMFONY_VERSION=3.4.*
|
||||
- COVERALLS_ENABLED="true"
|
||||
- PHPUNIT_FLAGS="-v --coverage-text --coverage-clover var/build/clover.xml"
|
||||
|
||||
@@ -56,51 +47,43 @@ matrix:
|
||||
- SYMFONY_VERSION=3.4.*
|
||||
- php: '7.1'
|
||||
env:
|
||||
- SYMFONY_VERSION=4.3.*
|
||||
- SYMFONY_VERSION=4.2.*
|
||||
- php: '7.1'
|
||||
env:
|
||||
- SYMFONY_VERSION=4.4.*
|
||||
- php: '7.2'
|
||||
env:
|
||||
- COVERALLS_ENABLED="true"
|
||||
- PHPUNIT_FLAGS="-v --coverage-text --coverage-clover var/build/clover.xml"
|
||||
- SYMFONY_VERSION=5.0.*
|
||||
- SYMFONY_VERSION=4.3.*
|
||||
|
||||
# Test unsupported versions of Symfony
|
||||
- php: '7.1'
|
||||
env:
|
||||
- SYMFONY_VERSION=4.1.*
|
||||
- SYMFONY_VERSION=4.0.*
|
||||
- php: '7.1'
|
||||
env:
|
||||
- SYMFONY_VERSION=4.2.*
|
||||
- SYMFONY_VERSION=4.1.*
|
||||
|
||||
|
||||
# Test upcoming Symfony versions with lowest supported PHP version and dev dependencies
|
||||
- php: '7.2'
|
||||
- php: '7.1'
|
||||
env:
|
||||
- STABILITY=dev
|
||||
- SYMFONY_VERSION=5.1.*
|
||||
- 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.*
|
||||
- SYMFONY_VERSION=4.4.*
|
||||
|
||||
before_install:
|
||||
- if [[ "$SYMFONY_VERSION" != "" ]]; then
|
||||
@@ -125,4 +108,4 @@ 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;
|
||||
46
CHANGELOG.md
46
CHANGELOG.md
@@ -1,46 +0,0 @@
|
||||
# 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
|
||||
249
README.md
249
README.md
@@ -3,7 +3,7 @@
|
||||
DataflowBundle is a bundle for Symfony 3.4+
|
||||
providing an easy way to create import / export dataflow.
|
||||
|
||||
[](https://travis-ci.com/code-rhapsodie/dataflow-bundle)
|
||||
[](https://travis-ci.org/code-rhapsodie/dataflow-bundle)
|
||||
|
||||
[](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
|
||||
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -1,3 +0,0 @@
|
||||
# Upgrade from v1.x to v2.0
|
||||
|
||||
[BC] `JobRepository` and `ScheduledDataflowRepository` are no longer a Doctrine ORM repository.
|
||||
@@ -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": [
|
||||
{
|
||||
@@ -42,22 +37,21 @@
|
||||
},
|
||||
"require": {
|
||||
"php": "^7.1",
|
||||
"ext-json": "*",
|
||||
"doctrine/dbal": "^2.0",
|
||||
"doctrine/doctrine-bundle": "^1.0||^2.0",
|
||||
"psr/log": "^1.1",
|
||||
"symfony/config": "^3.4||^4.0||^5.0",
|
||||
"symfony/console": "^3.4||^4.0||^5.0",
|
||||
"symfony/dependency-injection": "^3.4||>=4.1.12||^5.0",
|
||||
"symfony/event-dispatcher": "^3.4||^4.0||^5.0",
|
||||
"symfony/http-kernel": "^3.4||^4.0||^5.0",
|
||||
"symfony/lock": "^3.4||^4.0||^5.0",
|
||||
"symfony/monolog-bridge": "^3.4||^4.0||^5.0",
|
||||
"symfony/options-resolver": "^3.4||^4.0||^5.0",
|
||||
"symfony/validator": "^3.4||^4.0||^5.0",
|
||||
"symfony/yaml": "^3.4||^4.0||^5.0"
|
||||
"doctrine/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"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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')));
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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');
|
||||
|
||||
@@ -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();
|
||||
|
||||
|
||||
@@ -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') : '-',
|
||||
];
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
@@ -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]);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -13,6 +13,8 @@ interface DataflowInterface
|
||||
{
|
||||
/**
|
||||
* Processes the data.
|
||||
*
|
||||
* @return Result
|
||||
*/
|
||||
public function process(): Result;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
@@ -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']);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)]);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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
|
||||
{
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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)
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
{
|
||||
}
|
||||
@@ -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));
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -4,174 +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\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' => \PDO::PARAM_INT,
|
||||
'status' => \PDO::PARAM_INT,
|
||||
'label' => \PDO::PARAM_STR,
|
||||
'dataflow_type' => \PDO::PARAM_STR,
|
||||
'options' => \PDO::PARAM_STR,
|
||||
'requested_date' => 'datetime',
|
||||
'scheduled_dataflow_id' => \PDO::PARAM_INT,
|
||||
'count' => \PDO::PARAM_INT,
|
||||
'exceptions' => \PDO::PARAM_STR,
|
||||
'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, \PDO::PARAM_INT)))
|
||||
;
|
||||
|
||||
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, \PDO::PARAM_INT)));
|
||||
$stmt = $qb->execute();
|
||||
if (0 === $stmt->rowCount()) {
|
||||
return [];
|
||||
}
|
||||
while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
|
||||
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(), \PDO::PARAM_INT)))
|
||||
->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, \PDO::PARAM_INT)));
|
||||
|
||||
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, \PDO::PARAM_INT)))
|
||||
->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, \PDO::PARAM_INT)))
|
||||
->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->fetch(\PDO::FETCH_ASSOC))) {
|
||||
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, \PDO::PARAM_INT)))
|
||||
->orderBy('requested_date', 'DESC')
|
||||
->setMaxResults(20);
|
||||
$stmt = $qb->execute();
|
||||
if (0 === $stmt->rowCount()) {
|
||||
return [];
|
||||
}
|
||||
while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
|
||||
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->fetch(\PDO::FETCH_ASSOC))));
|
||||
$this->_em->persist($job);
|
||||
$this->_em->flush();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,42 +5,16 @@ declare(strict_types=1);
|
||||
namespace CodeRhapsodie\DataflowBundle\Repository;
|
||||
|
||||
use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
|
||||
use Doctrine\DBAL\Driver\Connection;
|
||||
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' => \PDO::PARAM_INT,
|
||||
'label' => \PDO::PARAM_STR,
|
||||
'dataflow_type' => \PDO::PARAM_STR,
|
||||
'options' => \PDO::PARAM_STR,
|
||||
'frequency' => \PDO::PARAM_STR,
|
||||
'next' => 'datetime',
|
||||
'enabled' => \PDO::PARAM_BOOL,
|
||||
];
|
||||
/**
|
||||
* @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.
|
||||
*
|
||||
@@ -48,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->fetch(\PDO::FETCH_ASSOC))) {
|
||||
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, \PDO::PARAM_INT)))
|
||||
->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->fetch(\PDO::FETCH_ASSOC))) {
|
||||
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()->fetchAll(\PDO::FETCH_ASSOC);
|
||||
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->fetch(\PDO::FETCH_ASSOC))));
|
||||
$this->_em->remove($dataflow);
|
||||
$this->_em->flush();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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'
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user