1 Commits
3.0.x ... v1.x

Author SHA1 Message Date
jbcr
4c80b3c51b add depentency to doctrine/doctrine-bundle (#29) 2019-11-08 14:58:46 +01:00
60 changed files with 693 additions and 2765 deletions

View File

@@ -23,91 +23,67 @@ env:
matrix:
fast_finish: true
include:
- php: '7.1'
- php: '7.2'
- php: '7.3'
- php: '7.4'
- php: '8.0'
# Enable code coverage with the previous supported PHP version
# - php: '7.4'
# env:
# - SYMFONY_VERSION=3.4.*
# - COVERALLS_ENABLED="true"
# - PHPUNIT_FLAGS="-v --coverage-text --coverage-clover var/build/clover.xml"
# Enable code coverage with the latest supported PHP version
# - php: '8.0'
# env:
# - SYMFONY_VERSION=3.4.*
# - COVERALLS_ENABLED="true"
# - PHPUNIT_FLAGS="-v --coverage-text --coverage-clover var/build/clover.xml"
# Minimum supported dependencies with the latest and oldest supported PHP versions
- php: '7.3'
env:
- COMPOSER_FLAGS="--prefer-lowest"
# Incompatibility between lowest symfony testing utils and phpunit
# - php: '8.0'
# env:
# - COMPOSER_FLAGS="--prefer-lowest"
# Test each supported Symfony version with lowest supported PHP version
# - php: '7.3'
# env:
# - SYMFONY_VERSION=3.4.*
- php: '7.3'
env:
- SYMFONY_VERSION=4.4.*
- php: '7.3'
env:
- COVERALLS_ENABLED="true"
- PHPUNIT_FLAGS="-v --coverage-text --coverage-clover var/build/clover.xml"
- SYMFONY_VERSION=5.2.*
# Minimum supported dependencies with the latest and oldest supported PHP versions
- php: '7.1'
env:
- COMPOSER_FLAGS="--prefer-lowest"
- php: '7.3'
env:
- COMPOSER_FLAGS="--prefer-lowest"
# Test each supported Symfony version with lowest supported PHP version
- php: '7.1'
env:
- SYMFONY_VERSION=3.4.*
- php: '7.1'
env:
- SYMFONY_VERSION=4.2.*
- php: '7.1'
env:
- SYMFONY_VERSION=4.3.*
# Test unsupported versions of Symfony
# - php: '7.3'
# env:
# - SYMFONY_VERSION=4.1.*
# - php: '7.3'
# env:
# - SYMFONY_VERSION=4.2.*
# - php: '7.3'
# env:
# - SYMFONY_VERSION=4.3.*
- php: '7.3'
- php: '7.1'
env:
- SYMFONY_VERSION=5.0.*
- php: '7.3'
- SYMFONY_VERSION=4.0.*
- php: '7.1'
env:
- SYMFONY_VERSION=5.1.*
- SYMFONY_VERSION=4.1.*
# Test upcoming Symfony versions with lowest supported PHP version and dev dependencies
# - php: '7.2'
# env:
# - STABILITY=dev
# - SYMFONY_VERSION=5.3.*
- php: '7.1'
env:
- STABILITY=dev
- SYMFONY_VERSION=4.4.*
# Test upcoming PHP versions with dev dependencies
#- php: '7.5snapshot'
# env:
# - STABILITY=dev
# - COMPOSER_FLAGS="--ignore-platform-reqs --prefer-stable"
- php: '7.4snapshot'
env:
- STABILITY=dev
- COMPOSER_FLAGS="--ignore-platform-reqs --prefer-stable"
allow_failures:
# 4.0 not supported because of https://github.com/advisories/GHSA-pgwj-prpq-jpc2
- env:
- SYMFONY_VERSION=4.0.*
- env:
- SYMFONY_VERSION=4.1.*
- env:
- SYMFONY_VERSION=4.2.*
- env:
- STABILITY=dev
- COMPOSER_FLAGS="--ignore-platform-reqs --prefer-stable"
- env:
- STABILITY=dev
- SYMFONY_VERSION=5.1.*
- env:
- STABILITY=dev
- SYMFONY_VERSION=5.2.*
- SYMFONY_VERSION=4.4.*
before_install:
- if [[ "$SYMFONY_VERSION" != "" ]]; then
@@ -121,15 +97,15 @@ before_install:
phpenv config-rm xdebug.ini || true;
fi
- if [[ "$COVERALLS_ENABLED" == "true" ]]; then
travis_retry composer require --dev php-coveralls/php-coveralls:^2.0 --no-update $COMPOSER_FLAGS;
travis_retry composer require --dev satooshi/php-coveralls:^2.0 --no-update $COMPOSER_FLAGS;
fi
install:
- travis_retry composer update --prefer-dist --no-interaction --no-progress --ansi $COMPOSER_FLAGS
- travis_retry composer update --prefer-dist --no-interaction --no-suggest --no-progress --ansi $COMPOSER_FLAGS
script: ./vendor/bin/phpunit $PHPUNIT_FLAGS
after_success:
- if [[ "$PHPUNIT_ENABLED" == "true" && "$COVERALLS_ENABLED" == "true" ]]; then
./vendor/bin/php-coveralls -vvv --config .coveralls.yml;
fi;
fi;

View File

@@ -1,58 +0,0 @@
# Version 3.1.0
* Added optional "messenger mode", to delegate jobs execution to workers from the Symfony messenger component
* Added support for asynchronous steps execution, using the AMPHP library (contribution from [matyo91](https://github.com/matyo91))
# Version 3.0.0
* Added PHP 8 support
* PHP minimum requirements bumped to 7.3
* Added Doctrine DBAL 3 support
* Doctrine DBAL minimum requirements bumped to 2.12
# Version 2.2.0
* Improve logging Dataflow job
# Version 2.1.1
* Fixed some Symfony 5 compatibility issues
# Version 2.1.0
* Added CollectionWriter and DelegatorWriter
* Adding Symfony 5.0 compatibility
* Save all exceptions caught in the log for `code-rhapsodie:dataflow:execute`
* Added more output when errors occured during `code-rhapsodie:dataflow:execute`
# Version 2.0.2
* Fixed the connection proxy class created by the factory
# Version 2.0.1
* Fixed next execution time not increasing for scheduled dataflows
# Version 2.0.0
* Add Doctrine DBAL multi-connection support
* Add configuration to define the default Doctrine DBAL connection
* Remove Doctrine ORM
* Rewrite repositories
# Version 1.0.1
* Fix lost dependency
* Fix schedule removing
# Version 1.0.0
Initial version
* Define and configure a Dataflow
* Run the Job scheduled
* Run one Dataflow from the command line
* Define the schedule for a Dataflow from the command line
* Enable/Disable a scheduled Dataflow from the command line
* Display the list of scheduled Dataflow from the command line
* Display the result for the last Job for a Dataflow from the command line

296
README.md
View File

@@ -3,13 +3,13 @@
DataflowBundle is a bundle for Symfony 3.4+
providing an easy way to create import / export dataflow.
[![Build Status](https://travis-ci.com/code-rhapsodie/dataflow-bundle.svg?branch=master)](https://travis-ci.com/code-rhapsodie/dataflow-bundle)
[![Build Status](https://travis-ci.org/code-rhapsodie/dataflow-bundle.svg?branch=master)](https://travis-ci.org/code-rhapsodie/dataflow-bundle)
[![Coverage Status](https://coveralls.io/repos/github/code-rhapsodie/dataflow-bundle/badge.svg)](https://coveralls.io/github/code-rhapsodie/dataflow-bundle)
Dataflow uses a linear generic workflow in three parts:
* one reader
* any number of steps that can be synchronous or asynchronous
* any number of steps
* one or more writers
The reader can read data from anywhere and return data row by row. Each step processes the current row data.
@@ -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,65 +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
```
### Messenger mode
Dataflow can delegate the execution of its jobs to the Symfony messenger component, if available.
This allows jobs to be executed concurrently by workers instead of sequentially.
To enable messenger mode:
```yaml
code_rhapsodie_dataflow:
messenger_mode:
enabled: true
# bus: 'messenger.default_bus' #Service ID of the bus you want Dataflow to use, if not the default one
```
You also need to route Dataflow messages to the proper transport:
```yaml
# config/packages/messenger.yaml
framework:
messenger:
transports:
async: '%env(MESSENGER_TRANSPORT_DSN)%'
routing:
CodeRhapsodie\DataflowBundle\MessengerMode\JobMessage: async
```
## Define a dataflow type
@@ -204,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;
})
@@ -218,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
@@ -241,7 +200,6 @@ If you're using Symfony auto-configuration for your services, this tag will be a
Otherwise, manually add the tag `coderhapsodie.dataflow.type` in your dataflow type service configuration:
```yaml
```yaml
CodeRhapsodie\DataflowExemple\DataflowType\MyFirstDataflowType:
tags:
@@ -264,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');
}
}
@@ -273,34 +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).
For asynchronous management, `AbstractDataflowType` come with two default options :
- loopInterval : default to 0. Update this interval if you wish customise the `tick` loop duration.
- emitInterval : default to 0. Update this interval to have a control when reader must emit new data in the flow pipeline.
### 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:
@@ -341,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);
}
}
}
@@ -370,7 +312,6 @@ $builder->setReader(($this->myReader)())
*Steps* are operations performed on the elements before they are handled by the *Writers*. Usually, steps are either:
- converters, that alter the element
- filters, that conditionally prevent further operations on the element
- generators, that can include asynchronous operations
A *Step* can be any callable, taking the element as its argument, and returning either:
- the element, possibly altered
@@ -379,26 +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;
});
// asynchronous step with 2 scale factor
$builder->addStep(function ($item): \Generator {
yield new \Amp\Delayed(1000); // asynchronous processing for 1 second long
// Titles are changed to all caps before export
$item['title'] = strtolower($item['title']);
return $item;
}, 2);
$builder->addStep(function ($item) {
$builder->addStep(function($item) {
// Private items are not exported
if ($item['private']) {
return false;
@@ -406,11 +335,8 @@ $builder->addStep(function ($item) {
return $item;
});
//[...]
```
Note : you can ensure writing order for asynchronous operations if all steps are scaled at 1 factor.
### Writers
*Writers* perform the actual import / export operations.
@@ -436,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.");
}
}
@@ -465,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.
@@ -570,8 +398,6 @@ Several commands are provided to manage schedules and run jobs.
`code-rhapsodie:dataflow:run-pending` Executes job in the queue according to their schedule.
When messenger mode is enabled, jobs will still be created according to their schedule, but execution will be handled by the messenger component instead.
`code-rhapsodie:dataflow:schedule:list` Display the list of dataflows scheduled.
`code-rhapsodie:dataflow:schedule:change-status` Enable or disable a scheduled dataflow
@@ -582,28 +408,6 @@ When messenger mode is enabled, jobs will still be created according to their sc
`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

View File

@@ -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);
}
};

View File

@@ -1,49 +0,0 @@
<?php
namespace CodeRhapsodie\DataflowBundle\Tests\DataflowType\Dataflow;
use Amp\Delayed;
use CodeRhapsodie\DataflowBundle\DataflowType\Dataflow\AMPAsyncDataflow;
use CodeRhapsodie\DataflowBundle\DataflowType\Dataflow\Dataflow;
use CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface;
use PHPUnit\Framework\TestCase;
class AMPAsyncDataflowTest extends TestCase
{
public function testProcess()
{
$reader = [1, 2, 3];
$result = [];
$dataflow = new AMPAsyncDataflow($reader, 'simple');
$dataflow->addStep(static function($item) {
return $item + 1;
});
$dataflow->addStep(static function($item): \Generator {
yield new Delayed(10); //delay 10 milliseconds
return $item * 2;
});
$dataflow->addWriter(new class($result) implements WriterInterface {
private $buffer;
public function __construct(&$buffer) {
$this->buffer = &$buffer;
}
public function prepare()
{
}
public function write($item)
{
$this->buffer[] = $item;
}
public function finish()
{
}
});
$dataflow->process();
self::assertSame([4, 6, 8], $result);
}
}

View File

@@ -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();
}
}

View File

@@ -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();
}
}

View File

@@ -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();
}
}

View File

@@ -1,58 +0,0 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\Tests\MessengerMode;
use CodeRhapsodie\DataflowBundle\Entity\Job;
use CodeRhapsodie\DataflowBundle\MessengerMode\JobMessage;
use CodeRhapsodie\DataflowBundle\MessengerMode\JobMessageHandler;
use CodeRhapsodie\DataflowBundle\Processor\JobProcessorInterface;
use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
use PHPUnit\Framework\MockObject\MockObject;
use PHPUnit\Framework\TestCase;
class JobMessageHandlerTest extends TestCase
{
/** @var JobRepository|MockObject */
private $repository;
/** @var JobProcessorInterface|MockObject */
private $processor;
/** @var JobMessageHandler */
private $handler;
protected function setUp(): void
{
$this->repository = $this->createMock(JobRepository::class);
$this->processor = $this->createMock(JobProcessorInterface::class);
$this->handler = new JobMessageHandler($this->repository, $this->processor);
}
public function testGetHandledMessages()
{
$this->assertSame([JobMessage::class], JobMessageHandler::getHandledMessages());
}
public function testInvoke()
{
$message = new JobMessage($id = 32);
$this->repository
->expects($this->once())
->method('find')
->with($id)
->willReturn($job = new Job())
;
$this->processor
->expects($this->once())
->method('process')
->with($job)
;
($this->handler)($message);
}
}

View File

@@ -1,120 +0,0 @@
<?php
namespace CodeRhapsodie\DataflowBundle\Tests\Processor;
use CodeRhapsodie\DataflowBundle\DataflowType\DataflowTypeInterface;
use CodeRhapsodie\DataflowBundle\DataflowType\Result;
use CodeRhapsodie\DataflowBundle\Entity\Job;
use CodeRhapsodie\DataflowBundle\Event\Events;
use CodeRhapsodie\DataflowBundle\Event\ProcessingEvent;
use CodeRhapsodie\DataflowBundle\Processor\JobProcessor;
use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
use PHPUnit\Framework\MockObject\MockObject;
use PHPUnit\Framework\TestCase;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;
class JobProcessorTest extends TestCase
{
/** @var JobProcessor */
private $processor;
/** @var JobRepository|MockObject */
private $repository;
/** @var DataflowTypeRegistryInterface|MockObject */
private $registry;
/** @var EventDispatcherInterface|MockObject */
private $dispatcher;
protected function setUp(): void
{
$this->repository = $this->createMock(JobRepository::class);
$this->registry = $this->createMock(DataflowTypeRegistryInterface::class);
$this->dispatcher = $this->createMock(EventDispatcherInterface::class);
$this->processor = new JobProcessor($this->repository, $this->registry, $this->dispatcher);
}
public function testProcess()
{
$now = new \DateTimeImmutable();
$job = (new Job())
->setStatus(Job::STATUS_PENDING)
->setDataflowType($type = 'type')
->setOptions($options = ['option1' => 'value1'])
;
// Symfony 3.4 to 4.4 call
if (!class_exists('Symfony\Contracts\EventDispatcher\Event')) {
$this->dispatcher
->expects($this->exactly(2))
->method('dispatch')
->withConsecutive(
[
Events::BEFORE_PROCESSING,
$this->callback(function (ProcessingEvent $event) use ($job) {
return $event->getJob() === $job;
})
],
[
Events::AFTER_PROCESSING,
$this->callback(function (ProcessingEvent $event) use ($job) {
return $event->getJob() === $job;
})
],
);
} else { // Symfony 5.0+
$this->dispatcher
->expects($this->exactly(2))
->method('dispatch')
->withConsecutive(
[
$this->callback(function (ProcessingEvent $event) use ($job) {
return $event->getJob() === $job;
}),
Events::BEFORE_PROCESSING,
],
[
$this->callback(function (ProcessingEvent $event) use ($job) {
return $event->getJob() === $job;
}),
Events::AFTER_PROCESSING,
],
);
}
$dataflowType = $this->createMock(DataflowTypeInterface::class);
$this->registry
->expects($this->once())
->method('getDataflowType')
->with($type)
->willReturn($dataflowType)
;
$bag = [new \Exception('message1')];
$result = new Result('name', new \DateTimeImmutable(), $end = new \DateTimeImmutable(), $count = 10, $bag);
$dataflowType
->expects($this->once())
->method('process')
->with($options)
->willReturn($result)
;
$this->repository
->expects($this->exactly(2))
->method('save')
;
$this->processor->process($job);
$this->assertGreaterThanOrEqual($now, $job->getStartTime());
$this->assertSame(Job::STATUS_COMPLETED, $job->getStatus());
$this->assertSame($end, $job->getEndTime());
$this->assertSame($count - count($bag), $job->getCount());
}
}

View File

@@ -1,72 +0,0 @@
<?php
namespace CodeRhapsodie\DataflowBundle\Tests\Runner;
use CodeRhapsodie\DataflowBundle\Entity\Job;
use CodeRhapsodie\DataflowBundle\MessengerMode\JobMessage;
use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
use CodeRhapsodie\DataflowBundle\Runner\MessengerDataflowRunner;
use PHPUnit\Framework\MockObject\MockObject;
use PHPUnit\Framework\TestCase;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\MessageBusInterface;
class MessengerDataflowRunnerTest extends TestCase
{
/** @var MessengerDataflowRunner */
private $runner;
/** @var JobRepository|MockObject */
private $repository;
/** @var MessageBusInterface|MockObject */
private $bus;
protected function setUp(): void
{
$this->repository = $this->createMock(JobRepository::class);
$this->bus = $this->createMock(MessageBusInterface::class);
$this->runner = new MessengerDataflowRunner($this->repository, $this->bus);
}
public function testRunPendingDataflows()
{
$job1 = (new Job())->setId($id1 = 10);
$job2 = (new Job())->setId($id2 = 20);
$this->repository
->expects($this->exactly(3))
->method('findNextPendingDataflow')
->willReturnOnConsecutiveCalls($job1, $job2, null)
;
$this->repository
->expects($this->exactly(2))
->method('save')
->withConsecutive([$job1], [$job2])
;
$this->bus
->expects($this->exactly(2))
->method('dispatch')
->withConsecutive([
$this->callback(function ($message) use ($id1) {
return $message instanceof JobMessage && $message->getJobId() === $id1;
})
], [
$this->callback(function ($message) use ($id2) {
return $message instanceof JobMessage && $message->getJobId() === $id2;
})
])
->willReturnOnConsecutiveCalls(
new Envelope(new JobMessage($id1)),
new Envelope(new JobMessage($id2))
)
;
$this->runner->runPendingDataflows();
$this->assertSame(Job::STATUS_QUEUED, $job1->getStatus());
$this->assertSame(Job::STATUS_QUEUED, $job2->getStatus());
}
}

View File

@@ -2,36 +2,59 @@
namespace CodeRhapsodie\DataflowBundle\Tests\Runner;
use CodeRhapsodie\DataflowBundle\DataflowType\DataflowTypeInterface;
use CodeRhapsodie\DataflowBundle\DataflowType\Result;
use CodeRhapsodie\DataflowBundle\Entity\Job;
use CodeRhapsodie\DataflowBundle\Processor\JobProcessorInterface;
use CodeRhapsodie\DataflowBundle\Event\Events;
use CodeRhapsodie\DataflowBundle\Event\ProcessingEvent;
use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
use CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunner;
use Doctrine\ORM\EntityManagerInterface;
use PHPUnit\Framework\MockObject\MockObject;
use PHPUnit\Framework\TestCase;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;
class PendingDataflowRunnerTest extends TestCase
{
/** @var PendingDataflowRunner */
private $runner;
/** @var EntityManagerInterface|MockObject */
private $em;
/** @var JobRepository|MockObject */
private $repository;
/** @var JobProcessorInterface|MockObject */
private $processor;
/** @var DataflowTypeRegistryInterface|MockObject */
private $registry;
/** @var EventDispatcherInterface|MockObject */
private $dispatcher;
protected function setUp(): void
{
$this->em = $this->createMock(EntityManagerInterface::class);
$this->repository = $this->createMock(JobRepository::class);
$this->processor = $this->createMock(JobProcessorInterface::class);
$this->registry = $this->createMock(DataflowTypeRegistryInterface::class);
$this->dispatcher = $this->createMock(EventDispatcherInterface::class);
$this->runner = new PendingDataflowRunner($this->repository, $this->processor);
$this->runner = new PendingDataflowRunner($this->em, $this->repository, $this->registry, $this->dispatcher);
}
public function testRunPendingDataflows()
{
$job1 = new Job();
$job2 = new Job();
$now = new \DateTime();
$job1 = (new Job())
->setStatus(Job::STATUS_PENDING)
->setDataflowType($type1 = 'type1')
->setOptions($options1 = ['option1' => 'value1'])
;
$job2 = (new Job())
->setStatus(Job::STATUS_PENDING)
->setDataflowType($type2 = 'type2')
->setOptions($options2 = ['option2' => 'value2'])
;
$this->repository
->expects($this->exactly(3))
@@ -39,12 +62,81 @@ class PendingDataflowRunnerTest extends TestCase
->willReturnOnConsecutiveCalls($job1, $job2, null)
;
$this->processor
$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);
$this->registry
->expects($this->exactly(2))
->method('getDataflowType')
->withConsecutive([$type1], [$type2])
->willReturnOnConsecutiveCalls($dataflowType1, $dataflowType2)
;
$bag1 = [new \Exception('message1')];
$bag2 = [new \Exception('message2')];
$result1 = new Result('name', new \DateTime(), $end1 = new \DateTime(), $count1 = 10, $bag1);
$result2 = new Result('name', new \DateTime(), $end2 = new \DateTime(), $count2 = 20, $bag2);
$dataflowType1
->expects($this->once())
->method('process')
->withConsecutive([$job1], [$job2])
->with($options1)
->willReturn($result1)
;
$dataflowType2
->expects($this->once())
->method('process')
->with($options2)
->willReturn($result2)
;
$this->em
->expects($this->exactly(4))
->method('flush')
;
$this->runner->runPendingDataflows();
$this->assertGreaterThanOrEqual($now, $job1->getStartTime());
$this->assertSame(Job::STATUS_COMPLETED, $job1->getStatus());
$this->assertSame($end1, $job1->getEndTime());
$this->assertSame($count1 - count($bag1), $job1->getCount());
$this->assertGreaterThanOrEqual($now, $job2->getStartTime());
$this->assertSame(Job::STATUS_COMPLETED, $job2->getStatus());
$this->assertSame($end2, $job2->getEndTime());
$this->assertSame($count2 - count($bag2), $job2->getCount());
}
}

View File

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

View File

@@ -2,12 +2,7 @@
"name": "code-rhapsodie/dataflow-bundle",
"description": "Data processing framework inspired by PortPHP",
"type": "symfony-bundle",
"keywords": [
"dataflow",
"import",
"export",
"data processing"
],
"keywords": ["dataflow", "import", "export", "data processing"],
"license": "MIT",
"authors": [
{
@@ -41,40 +36,33 @@
}
},
"require": {
"php": "^7.3||^8.0",
"ext-json": "*",
"doctrine/dbal": "^2.12||^3.0",
"doctrine/doctrine-bundle": "^1.0||^2.0",
"psr/log": "^1.1",
"symfony/config": "^3.4||^4.0||^5.0",
"symfony/console": "^3.4||^4.0||^5.0",
"symfony/dependency-injection": "^3.4||>=4.1.12||^5.0",
"symfony/event-dispatcher": "^3.4||^4.0||^5.0",
"symfony/http-kernel": "^3.4||^4.0||^5.0",
"symfony/lock": "^3.4||^4.0||^5.0",
"symfony/monolog-bridge": "^3.4||^4.0||^5.0",
"symfony/options-resolver": "^3.4||^4.0||^5.0",
"symfony/validator": "^3.4||^4.0||^5.0",
"symfony/yaml": "^3.4||^4.0||^5.0"
"php": "^7.1",
"doctrine/orm": "^2.4.5",
"seld/signal-handler": "^1.0",
"symfony/config": "^3.4||^4.0",
"symfony/console": "^3.4||^4.0",
"symfony/dependency-injection": "^3.4||^4.0",
"symfony/event-dispatcher": "^3.4||^4.0",
"symfony/http-kernel": "^3.4||^4.0",
"symfony/lock": "^3.4||^4.0",
"symfony/options-resolver": "^3.4||^4.0",
"symfony/validator": "^3.4||^4.0",
"symfony/yaml": "^3.4||^4.0",
"doctrine/doctrine-bundle": "^1.0"
},
"require-dev": {
"amphp/amp": "^2.5",
"phpunit/phpunit": "^7||^8||^9",
"symfony/messenger": "^4.4||^5.0"
"friendsofphp/php-cs-fixer": "^2.15",
"phpunit/phpunit": "^7||^8"
},
"suggest": {
"amphp/amp": "Provide asynchronous steps for your dataflows",
"portphp/portphp": "Provides generic readers, steps and writers for your dataflows.",
"symfony/messenger": "Allows messenger mode, i.e. letting workers run jobs"
"portphp/portphp": "Provides generic readers, steps and writers for your dataflows."
},
"config": {
"sort-packages": true
},
"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"
}
}
}

View File

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

View File

@@ -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;
}
}

View File

@@ -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')));

View File

@@ -4,40 +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 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;
}
/**
@@ -55,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');
;
}
/**
@@ -63,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;
}

View File

@@ -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');

View File

@@ -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();

View File

@@ -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') : '-',
];
}

View File

@@ -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;
}
}

View File

@@ -1,83 +0,0 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\DataflowType;
use CodeRhapsodie\DataflowBundle\DataflowType\Dataflow\AMPAsyncDataflow;
use CodeRhapsodie\DataflowBundle\DataflowType\Dataflow\DataflowInterface;
use CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface;
use Symfony\Component\OptionsResolver\OptionsResolver;
class AMPAsyncDataflowBuilder extends DataflowBuilder
{
/** @var int */
protected $loopInterval;
/** @var int */
protected $emitInterval;
public function __construct(?int $loopInterval = 0, ?int $emitInterval = 0)
{
$this->loopInterval = $loopInterval;
$this->emitInterval = $emitInterval;
}
/** @var string */
private $name;
/** @var iterable */
private $reader;
/** @var array */
private $steps = [];
/** @var WriterInterface[] */
private $writers = [];
public function setName(string $name): self
{
$this->name = $name;
return $this;
}
public function setReader(iterable $reader): self
{
$this->reader = $reader;
return $this;
}
public function addStep(callable $step, int $priority = 0, int $scale = 1): self
{
$this->steps[$priority][] = ['step' => $step, 'scale' => $scale];
return $this;
}
public function addWriter(WriterInterface $writer): self
{
$this->writers[] = $writer;
return $this;
}
public function getDataflow(): DataflowInterface
{
$dataflow = new AMPAsyncDataflow($this->reader, $this->name, $this->loopInterval, $this->emitInterval);
krsort($this->steps);
foreach ($this->steps as $stepArray) {
foreach ($stepArray as $step) {
$dataflow->addStep($step['step'], $step['scale']);
}
}
foreach ($this->writers as $writer) {
$dataflow->addWriter($writer);
}
return $dataflow;
}
}

View File

@@ -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
*/
@@ -27,22 +22,15 @@ abstract class AbstractDataflowType implements DataflowTypeInterface, LoggerAwar
$this->configureOptions($optionsResolver);
$options = $optionsResolver->resolve($options);
$builder = $this->createDataflowBuilder();
$builder->setName($this->getLabel());
$builder = (new DataflowBuilder())
->setName($this->getLabel())
;
$this->buildDataflow($builder, $options);
$dataflow = $builder->getDataflow();
if ($dataflow instanceof LoggerAwareInterface && $this->logger instanceof LoggerInterface) {
$dataflow->setLogger($this->logger);
}
return $dataflow->process();
}
protected function createDataflowBuilder(): DataflowBuilder
{
return new DataflowBuilder();
}
/**
* @codeCoverageIgnore
*/

View File

@@ -1,188 +0,0 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\DataflowType\Dataflow;
use function Amp\coroutine;
use Amp\Deferred;
use Amp\Delayed;
use Amp\Loop;
use Amp\Producer;
use Amp\Promise;
use function Amp\Promise\wait;
use CodeRhapsodie\DataflowBundle\DataflowType\Result;
use CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerAwareTrait;
use RuntimeException;
use Throwable;
class AMPAsyncDataflow implements DataflowInterface, LoggerAwareInterface
{
use LoggerAwareTrait;
/** @var string */
private $name;
/** @var iterable */
private $reader;
/** @var callable[] */
private $steps;
/** @var WriterInterface[] */
private $writers;
/** @var int */
private $loopInterval;
/** @var int */
private $emitInterval;
/** @var array */
private $states;
/** @var array */
private $stepsJobs;
public function __construct(iterable $reader, ?string $name, ?int $loopInterval = 0, ?int $emitInterval = 0)
{
$this->reader = $reader;
$this->name = $name;
$this->steps = [];
$this->writers = [];
$this->loopInterval = $loopInterval;
$this->emitInterval = $emitInterval;
$this->states = [];
$this->stepsJobs = [];
if (!function_exists('Amp\\Promise\\wait')) {
throw new RuntimeException('Amp is not loaded. Suggest install it with composer require amphp/amp');
}
}
/**
* @param int $scale
*
* @return $this
*/
public function addStep(callable $step, $scale = 1): self
{
$this->steps[] = [$step, $scale];
return $this;
}
/**
* @return $this
*/
public function addWriter(WriterInterface $writer): self
{
$this->writers[] = $writer;
return $this;
}
/**
* {@inheritdoc}
*/
public function process(): Result
{
$count = 0;
$exceptions = [];
$startTime = new \DateTimeImmutable();
try {
foreach ($this->writers as $writer) {
$writer->prepare();
}
$deferred = new Deferred();
$resolved = false; //missing $deferred->isResolved() in version 2.5
$producer = new Producer(function (callable $emit) {
foreach ($this->reader as $index => $item) {
yield new Delayed($this->emitInterval);
yield $emit([$index, $item]);
}
});
$watcherId = Loop::repeat($this->loopInterval, function () use ($deferred, &$resolved, $producer, &$count, &$exceptions) {
if (yield $producer->advance()) {
$it = $producer->getCurrent();
[$index, $item] = $it;
$this->states[$index] = [$index, 0, $item];
} elseif (!$resolved && 0 === count($this->states)) {
$resolved = true;
$deferred->resolve();
}
foreach ($this->states as $state) {
$this->processState($state, $count, $exceptions);
}
});
wait($deferred->promise());
Loop::cancel($watcherId);
foreach ($this->writers as $writer) {
$writer->finish();
}
} catch (\Throwable $e) {
$exceptions[] = $e;
$this->logException($e);
}
return new Result($this->name, $startTime, new \DateTimeImmutable(), $count, $exceptions);
}
/**
* @param mixed $state
* @param int $count internal count reference
* @param array $exceptions internal exceptions
*/
private function processState($state, int &$count, array &$exceptions): void
{
[$readIndex, $stepIndex, $item] = $state;
if ($stepIndex < count($this->steps)) {
if (!isset($this->stepsJobs[$stepIndex])) {
$this->stepsJobs[$stepIndex] = [];
}
[$step, $scale] = $this->steps[$stepIndex];
if (count($this->stepsJobs[$stepIndex]) < $scale && !isset($this->stepsJobs[$stepIndex][$readIndex])) {
$this->stepsJobs[$stepIndex][$readIndex] = true;
/** @var Promise<void> $promise */
$promise = coroutine($step)($item);
$promise->onResolve(function (?Throwable $exception = null, $newItem = null) use ($stepIndex, $readIndex, &$exceptions) {
if ($exception) {
$exceptions[$stepIndex] = $exception;
$this->logException($exception, (string) $stepIndex);
} elseif (false === $newItem) {
unset($this->states[$readIndex]);
} else {
$this->states[$readIndex] = [$readIndex, $stepIndex + 1, $newItem];
}
unset($this->stepsJobs[$stepIndex][$readIndex]);
});
}
} else {
unset($this->states[$readIndex]);
foreach ($this->writers as $writer) {
$writer->write($item);
}
++$count;
}
}
private function logException(Throwable $e, ?string $index = null): void
{
if (!isset($this->logger)) {
return;
}
$this->logger->error($e, ['exception' => $e, 'index' => $index]);
}
}

View File

@@ -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]);
}
}

View File

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

View File

@@ -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;

View File

@@ -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);
}
}

View File

@@ -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;
}

View File

@@ -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;
}
}

View File

@@ -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();
}

View File

@@ -24,15 +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']);
if ($config['messenger_mode']['enabled']) {
$container->setParameter('coderhapsodie.dataflow.bus', $config['messenger_mode']['bus']);
$loader->load('messenger_services.yaml');
}
}
}

View File

@@ -1,36 +0,0 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\DependencyInjection\Compiler;
use CodeRhapsodie\DataflowBundle\Runner\MessengerDataflowRunner;
use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
use Symfony\Component\DependencyInjection\ContainerBuilder;
use Symfony\Component\DependencyInjection\Exception\InvalidArgumentException;
use Symfony\Component\DependencyInjection\Reference;
class BusCompilerPass implements CompilerPassInterface
{
/**
* {@inheritdoc}
*/
public function process(ContainerBuilder $container)
{
if (!$container->hasParameter('coderhapsodie.dataflow.bus')) {
return;
}
$bus = $container->getParameter('coderhapsodie.dataflow.bus');
if (!$container->has($bus)) {
throw new InvalidArgumentException(sprintf('Service "%s" not found', $bus));
}
if (!$container->has(MessengerDataflowRunner::class)) {
return;
}
$definition = $container->findDefinition(MessengerDataflowRunner::class);
$definition->setArgument('$bus', new Reference($bus));
}
}

View File

@@ -1,34 +0,0 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\DependencyInjection\Compiler;
use CodeRhapsodie\DataflowBundle\Command\ExecuteDataflowCommand;
use CodeRhapsodie\DataflowBundle\Processor\JobProcessor;
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, JobProcessor::class] as $serviceId) {
if (!$container->has($serviceId)) {
continue;
}
$definition = $container->findDefinition($serviceId);
$definition->addMethodCall('setLogger', [new Reference($defaultLogger)]);
}
}
}

View File

@@ -1,51 +0,0 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\DependencyInjection;
use Symfony\Component\Config\Definition\Builder\TreeBuilder;
use Symfony\Component\Config\Definition\ConfigurationInterface;
use Symfony\Component\Messenger\MessageBusInterface;
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()
->arrayNode('messenger_mode')
->addDefaultsIfNotSet()
->children()
->booleanNode('enabled')
->defaultFalse()
->end()
->scalarNode('bus')
->defaultValue('messenger.default_bus')
->end()
->end()
->validate()
->ifTrue(static function ($v): bool { return $v['enabled'] && !interface_exists(MessageBusInterface::class); })
->thenInvalid('You need "symfony/messenger" in order to use Dataflow messenger mode.')
->end()
->end()
->end()
;
return $treeBuilder;
}
}

View File

@@ -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
@@ -16,88 +19,90 @@ class Job
const STATUS_PENDING = 0;
const STATUS_RUNNING = 1;
const STATUS_COMPLETED = 2;
const STATUS_QUEUED = 3;
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
@@ -108,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;
@@ -181,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;
@@ -193,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;
@@ -205,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;
@@ -217,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;
@@ -229,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;
@@ -253,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;
@@ -265,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;
@@ -277,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;

View File

@@ -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;

View File

@@ -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
{
}
}

View File

@@ -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;

View File

@@ -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)
));
}
}

View File

@@ -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
{
}

View File

@@ -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));
}
}

View File

@@ -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);
}
}

View File

@@ -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);
}
}
}

View File

@@ -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));
}
}

View File

@@ -1,21 +0,0 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\MessengerMode;
class JobMessage
{
/** @var int */
private $jobId;
public function __construct(int $jobId)
{
$this->jobId = $jobId;
}
public function getJobId(): int
{
return $this->jobId;
}
}

View File

@@ -1,34 +0,0 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\MessengerMode;
use CodeRhapsodie\DataflowBundle\Processor\JobProcessorInterface;
use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;
class JobMessageHandler implements MessageSubscriberInterface
{
/** @var JobRepository */
private $repository;
/** @var JobProcessorInterface */
private $processor;
public function __construct(JobRepository $repository, JobProcessorInterface $processor)
{
$this->repository = $repository;
$this->processor = $processor;
}
public function __invoke(JobMessage $message)
{
$this->processor->process($this->repository->find($message->getJobId()));
}
public static function getHandledMessages(): iterable
{
return [JobMessage::class];
}
}

View File

@@ -1,99 +0,0 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\Processor;
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 Symfony\Component\EventDispatcher\EventDispatcherInterface;
class JobProcessor implements JobProcessorInterface, LoggerAwareInterface
{
use LoggerAwareTrait;
/** @var JobRepository */
private $repository;
/** @var DataflowTypeRegistryInterface */
private $registry;
/** @var EventDispatcherInterface */
private $dispatcher;
public function __construct(JobRepository $repository, DataflowTypeRegistryInterface $registry, EventDispatcherInterface $dispatcher)
{
$this->repository = $repository;
$this->registry = $registry;
$this->dispatcher = $dispatcher;
}
public function process(Job $job): void
{
$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);
}
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);
}
$job
->setStatus(Job::STATUS_RUNNING)
->setStartTime(new \DateTime())
;
$this->repository->save($job);
}
private function afterProcessing(Job $job, Result $result, BufferHandler $bufferLogger): void
{
$job
->setEndTime($result->getEndTime())
->setStatus(Job::STATUS_COMPLETED)
->setCount($result->getSuccessCount())
->setExceptions($bufferLogger->clearBuffer())
;
$this->repository->save($job);
// Symfony 3.4 to 4.4 call
if (!class_exists('Symfony\Contracts\EventDispatcher\Event')) {
$this->dispatcher->dispatch(Events::AFTER_PROCESSING, new ProcessingEvent($job));
} else { // Symfony 5.0+ call
$this->dispatcher->dispatch(new ProcessingEvent($job), Events::AFTER_PROCESSING);
}
}
}

View File

@@ -1,12 +0,0 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\Processor;
use CodeRhapsodie\DataflowBundle\Entity\Job;
interface JobProcessorInterface
{
public function process(Job $job): void;
}

View File

@@ -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();
}
/**

View File

@@ -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;
}

View File

@@ -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;
}
}

View File

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

View File

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

View File

@@ -1,12 +0,0 @@
services:
CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunnerInterface: '@CodeRhapsodie\DataflowBundle\Runner\MessengerDataflowRunner'
CodeRhapsodie\DataflowBundle\Runner\MessengerDataflowRunner:
arguments:
$repository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
$bus: ~ # Filled in compiler pass
CodeRhapsodie\DataflowBundle\MessengerMode\JobMessageHandler:
arguments:
$repository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
$processor: '@CodeRhapsodie\DataflowBundle\Processor\JobProcessorInterface'
tags: ['messenger.message_handler']

View File

@@ -10,79 +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:
$repository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
$processor: '@CodeRhapsodie\DataflowBundle\Processor\JobProcessorInterface'
CodeRhapsodie\DataflowBundle\Processor\JobProcessorInterface: '@CodeRhapsodie\DataflowBundle\Processor\JobProcessor'
CodeRhapsodie\DataflowBundle\Processor\JobProcessor:
arguments:
$em: '@doctrine.orm.default_entity_manager'
$repository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
$registry: '@CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface'
$dispatcher: '@event_dispatcher'

View File

@@ -1,34 +0,0 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\Runner;
use CodeRhapsodie\DataflowBundle\Entity\Job;
use CodeRhapsodie\DataflowBundle\MessengerMode\JobMessage;
use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
use Symfony\Component\Messenger\MessageBusInterface;
class MessengerDataflowRunner implements PendingDataflowRunnerInterface
{
/** @var JobRepository */
private $repository;
/** @var MessageBusInterface */
private $bus;
public function __construct(JobRepository $repository, MessageBusInterface $bus)
{
$this->repository = $repository;
$this->bus = $bus;
}
public function runPendingDataflows(): void
{
while (null !== ($job = $this->repository->findNextPendingDataflow())) {
$this->bus->dispatch(new JobMessage($job->getId()));
$job->setStatus(Job::STATUS_QUEUED);
$this->repository->save($job);
}
}
}

View File

@@ -4,21 +4,35 @@ declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\Runner;
use CodeRhapsodie\DataflowBundle\Processor\JobProcessorInterface;
use CodeRhapsodie\DataflowBundle\DataflowType\Result;
use CodeRhapsodie\DataflowBundle\Entity\Job;
use CodeRhapsodie\DataflowBundle\Event\Events;
use CodeRhapsodie\DataflowBundle\Event\ProcessingEvent;
use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
use Doctrine\ORM\EntityManagerInterface;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;
class PendingDataflowRunner implements PendingDataflowRunnerInterface
{
/** @var EntityManagerInterface */
private $em;
/** @var JobRepository */
private $repository;
/** @var JobProcessorInterface */
private $processor;
/** @var DataflowTypeRegistryInterface */
private $registry;
public function __construct(JobRepository $repository, JobProcessorInterface $processor)
/** @var EventDispatcherInterface */
private $dispatcher;
public function __construct(EntityManagerInterface $em, JobRepository $repository, DataflowTypeRegistryInterface $registry, EventDispatcherInterface $dispatcher)
{
$this->em = $em;
$this->repository = $repository;
$this->processor = $processor;
$this->registry = $registry;
$this->dispatcher = $dispatcher;
}
/**
@@ -27,7 +41,49 @@ class PendingDataflowRunner implements PendingDataflowRunnerInterface
public function runPendingDataflows(): void
{
while (null !== ($job = $this->repository->findNextPendingDataflow())) {
$this->processor->process($job);
$this->beforeProcessing($job);
$dataflowType = $this->registry->getDataflowType($job->getDataflowType());
$result = $dataflowType->process($job->getOptions());
$this->afterProcessing($job, $result);
}
}
/**
* @param Job $job
*/
private function beforeProcessing(Job $job): void
{
$this->dispatcher->dispatch(Events::BEFORE_PROCESSING, new ProcessingEvent($job));
$job
->setStatus(Job::STATUS_RUNNING)
->setStartTime(new \DateTime())
;
$this->em->flush();
}
/**
* @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($exceptions)
;
$this->em->flush();
$this->dispatcher->dispatch(Events::AFTER_PROCESSING, new ProcessingEvent($job));
}
}

View File

@@ -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;
}
}