34 Commits

Author SHA1 Message Date
AUDUL
6d86ba16a0 SchemaDump command (#75)
* Added possibility to install/update database from command #37
2024-10-31 16:40:09 +01:00
AUDUL
95124acc26 * Fix compatibility with doctrine 4 (#73)
* Remove some deprecated
2024-10-31 15:38:10 +01:00
AUDUL
4011f39510 V5 (#72)
* Added Symfony 7 support

* Removed Symfony 6 compatibility

* Removed Symfony 5 compatibility

* Removed Symfony 4 compatibility

* Removed Symfony 3 compatibility

* Changed README.md

* Added CI

---------

Co-authored-by: Jérémy J <jeremy@code-rhapsodie.fr>
2024-10-31 13:01:43 +01:00
Matt Mankins
db37c4bdd1 Update Kudos Github Action to support generation from source repo only (#71)
* Update semicolons-kudos.yaml

* Update GitHub Action workflow for Semicolons Kudos Action

* Update semicolons-kudos.yaml

* Update GitHub Action workflow for Semicolons Kudos Action

* Update GitHub Action workflow for Semicolons Kudos Action

---------

Co-authored-by: semicolons-for-kudos[bot] <145267638+semicolons-for-kudos[bot]@users.noreply.github.com>
2023-12-27 17:29:46 +01:00
Olivier PORTIER
f20cd96ec5 Update semicolons-kudos.yaml 2023-12-20 11:49:50 +01:00
Olivier PORTIER
fd2c6aaab5 Update semicolons-kudos.yaml (#70) 2023-12-20 11:36:51 +01:00
Olivier PORTIER
4efd310a6e Initiate Kudos on dataflow-bundle by creating new file semicolons-kudos.yaml (#69) 2023-12-20 11:11:26 +01:00
Jérémy J
cec42a3337 Fix log exception argument typing 2023-12-06 13:56:16 +01:00
jbcr
d440ad008b add sonar config 2023-11-16 16:50:24 +01:00
jeremycr
e8b362526a Fix DBAL 2.12 compatibility break (#68) 2023-07-27 16:47:50 +02:00
jeremycr
3c56a90a93 Added the possibility to define a custom item index for exception logs (#66) 2023-07-27 09:38:22 +02:00
jeremycr
1b2b1be958 Removed travis for now, to be replaced by github actions in the future (#67) 2023-07-27 09:29:34 +02:00
jeremycr
25b2e9ec0f Upgrade for Symfony 6 (#65)
* Upgrade for Symfony 6
2022-08-18 09:36:43 +02:00
jeremycr
d89d7b72b0 Updated changelog for 3.1.0 (#64) 2021-04-21 10:40:27 +02:00
jeremycr
3cf05b555d Added messenger mode (#63) 2021-04-21 09:35:39 +02:00
Mathieu Ledru
9624f68675 Introduce the possibility to add asynchronous steps (#61) 2021-04-21 09:20:48 +02:00
jeremycr
b191e33c47 Added PHP 8 support and DBAL 3 support (#59) 2021-03-23 09:06:04 +01:00
jbcr
8bb55b1303 Update CHANGELOG.md (#56) 2021-01-15 15:18:36 +01:00
jeremycr
f0459462f7 Improved logging (#55)
* Improved logging

* Better handling of default logger

* Update src/DataflowType/Dataflow/Dataflow.php

* Updated README

Co-authored-by: jbcr <51637606+jbcr@users.noreply.github.com>
2021-01-15 14:38:48 +01:00
jeremycr
5a76c11bc6 Changelog for v2.1.1 (#54) 2020-12-02 15:44:23 +01:00
mdavid1297
d7efd85c8e Fix bug DateTime Oneshot (#53)
Co-authored-by: Marc DAVID <marc@Mac.local>
2020-12-02 15:13:27 +01:00
Jean-Baptiste Nahan
b0d17c31cc Fix some error with Symfony 5 and Symfoy 3.4 (#52)
* add Symfony 5 configuration initialisation with backward compatibility

* fix return value at end of command
2020-09-16 14:23:40 +02:00
Arnaud Lafon
e72d0d5e8d Bumped sf/di requirements to 4.1.12 minimum when 4.x used (#50) 2020-04-01 11:21:00 +02:00
jeremycr
a5518c80e2 Added more output when errors occured during execute command (#49)
* Added more output when errors occured during execute command

* Apply suggestions from code review

Co-Authored-By: jbcr <51637606+jbcr@users.noreply.github.com>
2020-02-27 09:57:42 +01:00
jeremycr
bd9171ad53 Added semi-colon after each query to ease copy-paste (#48) 2020-02-14 14:37:18 +01:00
jeremycr
5b4b3f1b6f Fixed the connection proxy class created by the factory (#47) 2020-01-30 14:37:47 +01:00
jbcr
d1330ae638 display errors catched during processing (#44)
* display errors catched during processing
* update changelog
2019-12-04 15:37:24 +01:00
jbcr
42b242ee6c catch all error (#43) 2019-12-04 15:37:00 +01:00
jeremycr
4d98adfe0a Updated composer aliases (#40) 2019-11-29 09:10:16 +01:00
jeremycr
c5fc6adf08 Passed PHP-CS-Fixer 2.16.1 (#39) 2019-11-28 14:38:16 +01:00
jeremycr
8efb4bd2d9 Removed unused dependency (#38) 2019-11-28 14:36:31 +01:00
jbcr
a9b19d933a Add Symfony 5.0 compatibility (#35)
* update travis config
* add doctrine dbal 2 in requirement
* remove php cs fixer. Use a global installed CS Fixer
* Add backward compatibility
* add coverage for SF5.0
* add tests
* code coverage on lowest version
* code coverage on 3.4 version
2019-11-26 14:25:05 +01:00
jeremycr
ca946429b1 Fixed next execution for scheduled dataflows not increasing (#36) 2019-11-22 14:12:04 +01:00
jeremycr
26ac98eb98 Added CollectionWriter and DelegatorWriter (#34)
* Added CollectionWriter and DelegatorWriter

* Added explanations and examples in README
2019-11-21 11:47:21 +01:00
75 changed files with 2226 additions and 1131 deletions

View File

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

27
.github/workflows/build.yml vendored Normal file
View File

@@ -0,0 +1,27 @@
name: Build
on:
push:
branches:
- master
jobs:
build:
name: Build
runs-on: ubuntu-latest
permissions: read-all
steps:
- uses: actions/checkout@v2
with:
fetch-depth: 0 # Shallow clones should be disabled for a better relevancy of analysis
- uses: sonarsource/sonarqube-scan-action@master
env:
SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
SONAR_HOST_URL: ${{ secrets.SONAR_HOST_URL }}
# If you wish to fail your job when the Quality Gate is red, uncomment the
# following lines. This would typically be used to fail a deployment.
# - uses: sonarsource/sonarqube-quality-gate-action@master
# timeout-minutes: 5
# env:
# SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}

12
.github/workflows/ci.yml vendored Normal file
View File

@@ -0,0 +1,12 @@
name: CI
on: [push]
jobs:
build-test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: php-actions/composer@v6 # or alternative dependency management
- uses: php-actions/phpunit@v4

26
.github/workflows/semicolons-kudos.yaml vendored Normal file
View File

@@ -0,0 +1,26 @@
name: Kudos for Code
on:
push:
branches: ["master"]
workflow_dispatch:
jobs:
kudos:
name: Semicolons Kudos
permissions: write-all
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0
- uses: LoremLabs/kudos-for-code-action@latest
with:
search-dir: "."
destination: "artifact"
generate-nomerges: true
generate-validemails: true
generate-limitdepth: 0
generate-fromrepo: true
analyze-repo: false
skip-ids: ""

3
.gitignore vendored
View File

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

View File

@@ -1,111 +0,0 @@
language: php
sudo: false
cache:
directories:
- $HOME/.composer/cache
branches:
only:
- master
- /^\d+\.\d+$/
- travis-setup
env:
global:
- SYMFONY_DEPRECATIONS_HELPER="max[self]=0"
- PHPUNIT_FLAGS="-v"
- PHPUNIT_ENABLED="true"
- STABILITY=stable
- COVERALLS_ENABLED="false"
matrix:
fast_finish: true
include:
- php: '7.1'
- php: '7.2'
- php: '7.3'
# Enable code coverage with the latest supported PHP version
- php: '7.3'
env:
- COVERALLS_ENABLED="true"
- PHPUNIT_FLAGS="-v --coverage-text --coverage-clover var/build/clover.xml"
# Minimum supported dependencies with the latest and oldest supported PHP versions
- php: '7.1'
env:
- COMPOSER_FLAGS="--prefer-lowest"
- php: '7.3'
env:
- COMPOSER_FLAGS="--prefer-lowest"
# 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.1'
env:
- SYMFONY_VERSION=4.0.*
- php: '7.1'
env:
- SYMFONY_VERSION=4.1.*
# Test upcoming Symfony versions with lowest supported PHP version and dev dependencies
- php: '7.1'
env:
- STABILITY=dev
- SYMFONY_VERSION=4.4.*
# Test upcoming PHP versions with dev dependencies
- php: '7.4snapshot'
env:
- STABILITY=dev
- COMPOSER_FLAGS="--ignore-platform-reqs --prefer-stable"
allow_failures:
- env:
- SYMFONY_VERSION=4.0.*
- env:
- SYMFONY_VERSION=4.1.*
- env:
- STABILITY=dev
- COMPOSER_FLAGS="--ignore-platform-reqs --prefer-stable"
- env:
- STABILITY=dev
- SYMFONY_VERSION=4.4.*
before_install:
- if [[ "$SYMFONY_VERSION" != "" ]]; then
travis_retry composer global require "symfony/flex:^1.4";
composer config extra.symfony.require $SYMFONY_VERSION;
fi
- if [[ "$STABILITY" != "stable" ]]; then
travis_retry composer config minimum-stability $STABILITY;
fi
- if [[ "$COVERALLS_ENABLED" != "true" ]]; then
phpenv config-rm xdebug.ini || true;
fi
- if [[ "$COVERALLS_ENABLED" == "true" ]]; then
travis_retry composer require --dev satooshi/php-coveralls:^2.0 --no-update $COMPOSER_FLAGS;
fi
install:
- travis_retry composer update --prefer-dist --no-interaction --no-suggest --no-progress --ansi $COMPOSER_FLAGS
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;

View File

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

268
README.md
View File

@@ -1,18 +1,23 @@
# Code Rhapsodie Dataflow Bundle
DataflowBundle is a bundle for Symfony 3.4+
DataflowBundle is a bundle for Symfony 3.4+
providing an easy way to create import / export dataflow.
[![Build Status](https://travis-ci.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 | Symfony | Support |
|----------|--------------------------|---------|
| 5.x | 7.x | yes |
| 4.x | 3.4 \| 4.x \| 5.x \| 6.x | yes |
| 3.x | 3.4 \| 4.x \| 5.x | no |
| 2.x | 3.4 \| 4.x | no |
| 1.x | 3.4 \| 4.x | no |
Dataflow uses a linear generic workflow in three parts:
* one reader
* any number of steps
* one or more writers
The reader can read data from anywhere and return data row by row. Each step processes the current row data.
* one reader
* any number of steps that can be synchronous or asynchronous
* one or more writers
The reader can read data from anywhere and return data row by row. Each step processes the current row data.
The steps are executed in the order in which they are added.
And, one or more writers save the row anywhere you want.
@@ -20,7 +25,6 @@ As the following schema shows, you can define more than one dataflow:
![Dataflow schema](src/Resources/doc/schema.png)
# Features
* Define and configure a Dataflow
@@ -32,9 +36,10 @@ As the following schema shows, you can define more than one dataflow:
* Display the result for the last Job for a Dataflow from the command line
* Work with multiple Doctrine DBAL connections
## Installation
Security notice: Symfony 4.x is not supported before 4.1.12, see https://github.com/advisories/GHSA-pgwj-prpq-jpc2
### Add the dependency
To install this bundle, run this command :
@@ -47,7 +52,8 @@ $ composer require code-rhapsodie/dataflow-bundle
You can use the generic readers, writers and steps from [PortPHP](https://github.com/portphp/portphp).
For the writers, you must use the adapter `CodeRhapsodie\DataflowBundle\DataflowType\Writer\PortWriterAdapter` like this:
For the writers, you must use the adapter `CodeRhapsodie\DataflowBundle\DataflowType\Writer\PortWriterAdapter` like
this:
```php
<?php
@@ -60,9 +66,7 @@ $builder->addWriter(new \CodeRhapsodie\DataflowBundle\DataflowType\Writer\PortWr
### Register the bundle
#### Symfony 4 (new tree)
For Symfony 4, add `CodeRhapsodie\DataflowBundle\CodeRhapsodieDataflowBundle::class => ['all' => true],
Add `CodeRhapsodie\DataflowBundle\CodeRhapsodieDataflowBundle::class => ['all' => true],
` in the `config/bundles.php` file.
Like this:
@@ -77,32 +81,13 @@ return [
];
```
#### Symfony 3.4 (old tree)
For Symfony 3.4, add a new line in the `app/AppKernel.php` file.
Like this:
```php
<?php
// app/AppKernel.php
public function registerBundles()
{
$bundles = [
// ...
new CodeRhapsodie\DataflowBundle\CodeRhapsodieDataflowBundle(),
// ...
];
}
```
### Update the database
This bundle uses Doctrine DBAL to store Dataflow schedule into the database table (`cr_dataflow_scheduled`)
and jobs (`cr_dataflow_job`).
If you use [Doctrine Migration Bundle](https://symfony.com/doc/master/bundles/DoctrineMigrationsBundle/index.html) or [Phinx](https://phinx.org/)
If you use [Doctrine Migration Bundle](https://symfony.com/doc/master/bundles/DoctrineMigrationsBundle/index.html)
or [Phinx](https://phinx.org/)
or [Kaliop Migration Bundle](https://github.com/kaliop-uk/ezmigrationbundle) or whatever,
you can add a new migration with the generated SQL query from this command:
@@ -126,11 +111,48 @@ 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
This bundle uses a fixed and simple workflow structure in order to let you focus on the data processing logic part of your dataflow.
This bundle uses a fixed and simple workflow structure in order to let you focus on the data processing logic part of
your dataflow.
A dataflow type defines the different parts of your dataflow. A dataflow is made of:
- exactly one *Reader*
- any number of *Steps*
- one or more *Writers*
@@ -139,8 +161,10 @@ Dataflow types can be configured with options.
A dataflow type must implement `CodeRhapsodie\DataflowBundle\DataflowType\DataflowTypeInterface`.
To help with creating your dataflow types, an abstract class `CodeRhapsodie\DataflowBundle\DataflowType\AbstractDataflowType`
is provided, allowing you to define your dataflow through a handy builder `CodeRhapsodie\DataflowBundle\DataflowType\DataflowBuilder`.
To help with creating your dataflow types, an abstract
class `CodeRhapsodie\DataflowBundle\DataflowType\AbstractDataflowType`
is provided, allowing you to define your dataflow through a handy
builder `CodeRhapsodie\DataflowBundle\DataflowType\DataflowBuilder`.
This is an example to define one class DataflowType:
@@ -200,14 +224,16 @@ class MyFirstDataflowType extends AbstractDataflowType
Dataflow types must be tagged with `coderhapsodie.dataflow.type`.
If you're using Symfony auto-configuration for your services, this tag will be automatically added to all services implementing `DataflowTypeInterface`.
If you're using Symfony auto-configuration for your services, this tag will be automatically added to all services
implementing `DataflowTypeInterface`.
Otherwise, manually add the tag `coderhapsodie.dataflow.type` in your dataflow type service configuration:
```yaml
CodeRhapsodie\DataflowExemple\DataflowType\MyFirstDataflowType:
tags:
- { name: coderhapsodie.dataflow.type }
```yaml
CodeRhapsodie\DataflowExemple\DataflowType\MyFirstDataflowType:
tags:
- { name: coderhapsodie.dataflow.type }
```
### Use options for your dataflow type
@@ -233,14 +259,46 @@ class MyFirstDataflowType extends AbstractDataflowType
}
```
With this configuration, the option `fileName` is required. For an advanced usage of the option resolver, read the [Symfony documentation](https://symfony.com/doc/current/components/options_resolver.html).
With this configuration, the option `fileName` is required. For an advanced usage of the option resolver, read
the [Symfony documentation](https://symfony.com/doc/current/components/options_resolver.html).
For asynchronous management, `AbstractDataflowType` come with two default options :
- loopInterval : default to 0. Update this interval if you wish customise the `tick` loop duration.
- emitInterval : default to 0. Update this interval to have a control when reader must emit new data in the flow
pipeline.
### 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:
```shell script
$ bin/console debug:container --tag coderhapsodie.dataflow.type --show-private
$ bin/console debug:container --tag coderhapsodie.dataflow.type
```
The result is like this:
@@ -257,10 +315,10 @@ Symfony Container Public and Private Services Tagged with "coderhapsodie.dataflo
```
### Readers
*Readers* provide the dataflow with elements to import / export. Usually, elements are read from an external resource (file, database, webservice, etc).
*Readers* provide the dataflow with elements to import / export. Usually, elements are read from an external resource (
file, database, webservice, etc).
A *Reader* can be any `iterable`.
@@ -298,14 +356,16 @@ You can set up this reader as follows:
$builder->setReader(($this->myReader)())
```
### Steps
*Steps* are operations performed on the elements before they are handled by the *Writers*. Usually, steps are either:
- converters, that alter the element
- filters, that conditionally prevent further operations on the element
- generators, that can include asynchronous operations
A *Step* can be any callable, taking the element as its argument, and returning either:
- the element, possibly altered
- `false`, if no further operations should be performed on this element
@@ -321,6 +381,16 @@ $builder->addStep(function ($item) {
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) {
// Private items are not exported
if ($item['private']) {
@@ -332,12 +402,15 @@ $builder->addStep(function ($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.
A *Writer* must implement `CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface`.
As this interface is not compatible with `Port\Writer`, the adapter `CodeRhapsodie\DataflowBundle\DataflowType\Writer\PortWriterAdapter` is provided.
As this interface is not compatible with `Port\Writer`, the
adapter `CodeRhapsodie\DataflowBundle\DataflowType\Writer\PortWriterAdapter` is provided.
This example show how to use the predefined PhpPort Writer :
@@ -386,6 +459,100 @@ 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.
@@ -402,6 +569,9 @@ 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
@@ -439,6 +609,8 @@ $ bin/console code-rhapsodie:dataflow:run-pending --connection=default
Please report issues and request features at https://github.com/code-rhapsodie/dataflow-bundle/issues.
Please note that only the last release of the 4.x and the 5.x versions of this bundle are actively supported.
# Contributing
Contributions are very welcome. Please see [CONTRIBUTING.md](CONTRIBUTING.md) for

View File

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

View File

@@ -0,0 +1,46 @@
<?php
namespace CodeRhapsodie\DataflowBundle\Tests\DataflowType\Dataflow;
use Amp\Delayed;
use CodeRhapsodie\DataflowBundle\DataflowType\Dataflow\AMPAsyncDataflow;
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 fn($item) => $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

@@ -0,0 +1,57 @@
<?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')
;
$matcher = $this->exactly(count($values));
$embeddedWriter
->expects($matcher)
->method('write')
->with($this->callback(function ($arg) use ($matcher, $values) {
return $arg === $values[$matcher->numberOfInvocations() - 1];
}))
;
$writer = new CollectionWriter($embeddedWriter);
$writer->prepare();
$writer->write($values);
$writer->finish();
}
}

View File

@@ -0,0 +1,147 @@
<?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
{
private DelegatorWriter $delegatorWriter;
private DelegateWriterInterface|MockObject $delegateInt;
private DelegateWriterInterface|MockObject $delegateString;
private DelegateWriterInterface|MockObject $delegateArray;
protected function setUp(): void
{
$this->delegateInt = $this->createMock(DelegateWriterInterface::class);
$this->delegateInt->method('supports')->willReturnCallback(fn($argument) => is_int($argument));
$this->delegateString = $this->createMock(DelegateWriterInterface::class);
$this->delegateString->method('supports')->willReturnCallback(fn($argument) => is_string($argument));
$this->delegateArray = $this->createMock(DelegateWriterInterface::class);
$this->delegateArray->method('supports')->willReturnCallback(fn($argument) => 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

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

View File

@@ -2,31 +2,21 @@
namespace CodeRhapsodie\DataflowBundle\Tests\Manager;
use CodeRhapsodie\DataflowBundle\DataflowType\DataflowTypeInterface;
use CodeRhapsodie\DataflowBundle\Entity\Job;
use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
use CodeRhapsodie\DataflowBundle\Exceptions\UnknownDataflowTypeException;
use CodeRhapsodie\DataflowBundle\Manager\ScheduledDataflowManager;
use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
use Doctrine\DBAL\Driver\Connection;
use Doctrine\ORM\EntityManagerInterface;
use Doctrine\DBAL\Connection;
use PHPUnit\Framework\MockObject\MockObject;
use PHPUnit\Framework\TestCase;
class ScheduledDataflowManagerTest extends TestCase
{
/** @var ScheduledDataflowManager */
private $manager;
/** @var Connection|MockObject */
private $connection;
/** @var ScheduledDataflowRepository|MockObject */
private $scheduledDataflowRepository;
/** @var JobRepository|MockObject */
private $jobRepository;
private ScheduledDataflowManager $manager;
private Connection|MockObject $connection;
private ScheduledDataflowRepository|MockObject $scheduledDataflowRepository;
private JobRepository|MockObject $jobRepository;
protected function setUp(): void
{
@@ -55,10 +45,20 @@ class ScheduledDataflowManagerTest extends TestCase
->willReturn([$scheduled1, $scheduled2])
;
$matcher = $this->exactly(2);
$this->jobRepository
->expects($this->exactly(2))
->expects($matcher)
->method('findPendingForScheduledDataflow')
->withConsecutive([$scheduled1], [$scheduled2])
->with($this->callback(function ($arg) use ($matcher, $scheduled1, $scheduled2) {
switch ($matcher->numberOfInvocations()) {
case 1:
return $arg === $scheduled1;
case 2:
return $arg === $scheduled2;
default:
return false;
}
}))
->willReturnOnConsecutiveCalls(new Job(), null)
;
@@ -70,16 +70,12 @@ class ScheduledDataflowManagerTest extends TestCase
->expects($this->once())
->method('save')
->with(
$this->callback(function (Job $job) use ($type, $options, $next, $label, $scheduled2) {
return (
$job->getStatus() === Job::STATUS_PENDING
&& $job->getDataflowType() === $type
&& $job->getOptions() === $options
&& $job->getRequestedDate() == $next
&& $job->getLabel() === $label
&& $job->getScheduledDataflowId() === $scheduled2->getId()
);
})
$this->callback(fn(Job $job) => $job->getStatus() === Job::STATUS_PENDING
&& $job->getDataflowType() === $type
&& $job->getOptions() === $options
&& $job->getRequestedDate() == $next
&& $job->getLabel() === $label
&& $job->getScheduledDataflowId() === $scheduled2->getId())
)
;
@@ -98,4 +94,44 @@ class ScheduledDataflowManagerTest extends TestCase
$this->assertEquals($next->add(\DateInterval::createFromDateString($frequency)), $scheduled2->getNext());
}
public function testCreateJobsFromScheduledDataflowsWithError()
{
$scheduled1 = new ScheduledDataflow();
$this->scheduledDataflowRepository
->expects($this->once())
->method('findReadyToRun')
->willReturn([$scheduled1])
;
$this->jobRepository
->expects($this->exactly(1))
->method('findPendingForScheduledDataflow')
->with($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

@@ -0,0 +1,48 @@
<?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
{
private JobRepository|MockObject $repository;
private JobProcessorInterface|MockObject $processor;
private JobMessageHandler $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 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

@@ -0,0 +1,94 @@
<?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
{
private JobProcessor $processor;
private JobRepository|MockObject $repository;
private DataflowTypeRegistryInterface|MockObject $registry;
private EventDispatcherInterface|MockObject $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'])
;
$matcher = $this->exactly(2);
$this->dispatcher
->expects($matcher)
->method('dispatch')
->with(
$this->callback(function ($arg) use ($job) {
return $arg instanceof ProcessingEvent && $arg->getJob() === $job;
}),
$this->callback(function ($arg) use ($matcher) {
switch ($matcher->numberOfInvocations()) {
case 1:
return $arg === Events::BEFORE_PROCESSING;
case 2:
return $arg === Events::AFTER_PROCESSING;
default:
return false;
}
})
);
$dataflowType = $this->createMock(DataflowTypeInterface::class);
$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

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

View File

@@ -0,0 +1,79 @@
<?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
{
private MessengerDataflowRunner $runner;
private JobRepository|MockObject $repository;
private MessageBusInterface|MockObject $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)
;
$matcher = $this->exactly(2);
$this->repository
->expects($matcher)
->method('save')
->with($this->callback(function ($arg) use ($matcher, $job1, $job2) {
switch ($matcher->numberOfInvocations()) {
case 1:
return $arg === $job1;
case 2:
return $arg === $job2;
default:
return false;
}
}))
;
$matcher = $this->exactly(2);
$this->bus
->expects($matcher)
->method('dispatch')
->with($this->callback(function ($arg) use ($matcher, $id1, $id2) {
switch ($matcher->numberOfInvocations()) {
case 1:
return $arg instanceof JobMessage && $arg->getJobId() === $id1;
case 2:
return $arg instanceof JobMessage && $arg->getJobId() === $id2;
default:
return false;
}
}))
->willReturnOnConsecutiveCalls(
new Envelope(new JobMessage($id1)),
new Envelope(new JobMessage($id2))
)
;
$this->runner->runPendingDataflows();
$this->assertSame(Job::STATUS_QUEUED, $job1->getStatus());
$this->assertSame(Job::STATUS_QUEUED, $job2->getStatus());
}
}

View File

@@ -2,55 +2,31 @@
namespace CodeRhapsodie\DataflowBundle\Tests\Runner;
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\Registry\DataflowTypeRegistryInterface;
use CodeRhapsodie\DataflowBundle\Processor\JobProcessorInterface;
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 JobRepository|MockObject */
private $repository;
/** @var DataflowTypeRegistryInterface|MockObject */
private $registry;
/** @var EventDispatcherInterface|MockObject */
private $dispatcher;
private PendingDataflowRunner $runner;
private JobRepository|MockObject $repository;
private JobProcessorInterface|MockObject $processor;
protected function setUp(): void
{
$this->repository = $this->createMock(JobRepository::class);
$this->registry = $this->createMock(DataflowTypeRegistryInterface::class);
$this->dispatcher = $this->createMock(EventDispatcherInterface::class);
$this->processor = $this->createMock(JobProcessorInterface::class);
$this->runner = new PendingDataflowRunner($this->repository, $this->registry, $this->dispatcher);
$this->runner = new PendingDataflowRunner($this->repository, $this->processor);
}
public function testRunPendingDataflows()
{
$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'])
;
$job1 = new Job();
$job2 = new Job();
$this->repository
->expects($this->exactly(3))
@@ -58,81 +34,22 @@ class PendingDataflowRunnerTest extends TestCase
->willReturnOnConsecutiveCalls($job1, $job2, null)
;
$this->dispatcher
->expects($this->exactly(4))
->method('dispatch')
->withConsecutive(
[
Events::BEFORE_PROCESSING,
$this->callback(function (ProcessingEvent $event) use ($job1) {
return $event->getJob() === $job1;
})
],
[
Events::AFTER_PROCESSING,
$this->callback(function (ProcessingEvent $event) use ($job1) {
return $event->getJob() === $job1;
})
],
[
Events::BEFORE_PROCESSING,
$this->callback(function (ProcessingEvent $event) use ($job2) {
return $event->getJob() === $job2;
})
],
[
Events::AFTER_PROCESSING,
$this->callback(function (ProcessingEvent $event) use ($job2) {
return $event->getJob() === $job2;
})
]
)
;
$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())
$matcher = $this->exactly(2);
$this->processor
->expects($matcher)
->method('process')
->with($options1)
->willReturn($result1)
;
$dataflowType2
->expects($this->once())
->method('process')
->with($options2)
->willReturn($result2)
;
$this->repository
->expects($this->exactly(4))
->method('save')
->with($this->callback(function ($arg) use ($matcher, $job1, $job2) {
switch ($matcher->numberOfInvocations()) {
case 1:
return $arg === $job1;
case 2:
return $arg === $job2;
default:
return false;
}
}))
;
$this->runner->runPendingDataflows();
$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

@@ -4,18 +4,18 @@ namespace CodeRhapsodie\DataflowBundle\Tests\Validator\Constraints;
use CodeRhapsodie\DataflowBundle\Validator\Constraints\Frequency;
use CodeRhapsodie\DataflowBundle\Validator\Constraints\FrequencyValidator;
use PHPUnit\Framework\Attributes\DataProvider;
use Symfony\Component\Validator\ConstraintValidatorInterface;
use Symfony\Component\Validator\Test\ConstraintValidatorTestCase;
class FrequencyValidatorTest extends ConstraintValidatorTestCase
{
protected function createValidator()
protected function createValidator(): ConstraintValidatorInterface
{
return new FrequencyValidator();
}
/**
* @dataProvider getValidValues
*/
#[DataProvider('getValidValues')]
public function testValidValues($value)
{
$this->validator->validate($value, new Frequency());
@@ -23,7 +23,7 @@ class FrequencyValidatorTest extends ConstraintValidatorTestCase
$this->assertNoViolation();
}
public function getValidValues()
public static function getValidValues()
{
return [
['3 days'],
@@ -47,9 +47,7 @@ class FrequencyValidatorTest extends ConstraintValidatorTestCase
;
}
/**
* @dataProvider getNegativeValues
*/
#[DataProvider('getNegativeValues')]
public function testNegativeIntervals($value)
{
$constraint = new Frequency([
@@ -64,7 +62,7 @@ class FrequencyValidatorTest extends ConstraintValidatorTestCase
;
}
public function getNegativeValues()
public static function getNegativeValues()
{
return [
['now'],

View File

@@ -2,7 +2,12 @@
"name": "code-rhapsodie/dataflow-bundle",
"description": "Data processing framework inspired by PortPHP",
"type": "symfony-bundle",
"keywords": ["dataflow", "import", "export", "data processing"],
"keywords": [
"dataflow",
"import",
"export",
"data processing"
],
"license": "MIT",
"authors": [
{
@@ -36,33 +41,42 @@
}
},
"require": {
"php": "^7.1",
"doctrine/dbal": "^2.0",
"seld/signal-handler": "^1.0",
"symfony/config": "^3.4||^4.0",
"symfony/console": "^3.4||^4.0",
"symfony/dependency-injection": "^3.4||^4.0",
"symfony/event-dispatcher": "^3.4||^4.0",
"symfony/http-kernel": "^3.4||^4.0",
"symfony/lock": "^3.4||^4.0",
"symfony/options-resolver": "^3.4||^4.0",
"symfony/validator": "^3.4||^4.0",
"symfony/yaml": "^3.4||^4.0",
"doctrine/doctrine-bundle": "^1.0"
"php": "^8.0",
"ext-json": "*",
"doctrine/dbal": "^3.0||^4.0",
"doctrine/doctrine-bundle": "^2.0",
"monolog/monolog": "^2.0||^3.0",
"psr/log": "^1.1||^2.0||^3.0",
"symfony/config": "^7.0",
"symfony/console": "^7.0",
"symfony/dependency-injection": "^7.0",
"symfony/event-dispatcher": "^7.0",
"symfony/http-kernel": "^7.0",
"symfony/lock": "^7.0",
"symfony/monolog-bridge": "^7.0",
"symfony/options-resolver": "^7.0",
"symfony/validator": "^7.0",
"symfony/yaml": "^7.0"
},
"require-dev": {
"friendsofphp/php-cs-fixer": "^2.15",
"phpunit/phpunit": "^7||^8"
"amphp/amp": "^2.5",
"phpunit/phpunit": "^11",
"portphp/portphp": "^1.9",
"rector/rector": "^1.0",
"symfony/messenger": "^7.0"
},
"suggest": {
"portphp/portphp": "Provides generic readers, steps and writers for your dataflows."
"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"
},
"config": {
"sort-packages": true
},
"extra": {
"branch-alias": {
"dev-master": "2.x-dev",
"dev-master": "2.1.x-dev",
"dev-v2.0.x": "2.0.x-dev",
"dev-v1.x": "1.x-dev"
}
}

View File

@@ -1,32 +1,21 @@
<?xml version = '1.0' encoding = 'UTF-8'?>
<?xml version="1.0" encoding="UTF-8"?>
<!-- http://www.phpunit.de/manual/current/en/appendixes.configuration.html -->
<phpunit
backupGlobals="false"
backupStaticAttributes="false"
bootstrap="Tests/bootstrap.php"
convertErrorsToExceptions="true"
convertNoticesToExceptions="true"
convertWarningsToExceptions="true"
colors="false"
>
<php>
<ini name="error_reporting" value="-1" />
</php>
<testsuites>
<testsuite name="Dataflow tests suite">
<directory suffix="Test.php">./Tests</directory>
</testsuite>
</testsuites>
<filter>
<whitelist>
<directory>./src/</directory>
<exclude>
<directory>Tests/</directory>
<directory>vendor/</directory>
</exclude>
</whitelist>
</filter>
<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" backupGlobals="false" bootstrap="Tests/bootstrap.php" colors="false" xsi:noNamespaceSchemaLocation="https://schema.phpunit.de/11.1/phpunit.xsd" cacheDirectory=".phpunit.cache" backupStaticProperties="false">
<php>
<ini name="error_reporting" value="-1"/>
</php>
<testsuites>
<testsuite name="Dataflow tests suite">
<directory suffix="Test.php">./Tests</directory>
</testsuite>
</testsuites>
<source>
<include>
<directory>./src/</directory>
</include>
<exclude>
<directory>Tests/</directory>
<directory>vendor/</directory>
</exclude>
</source>
</phpunit>

25
rector.php Normal file
View File

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

4
sonar-project.properties Normal file
View File

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

View File

@@ -5,8 +5,11 @@ 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\DependencyInjection\Extension\ExtensionInterface;
use Symfony\Component\HttpKernel\Bundle\Bundle;
/**
@@ -14,15 +17,19 @@ use Symfony\Component\HttpKernel\Bundle\Bundle;
*/
class CodeRhapsodieDataflowBundle extends Bundle
{
protected $name = 'CodeRhapsodieDataflowBundle';
protected string $name = 'CodeRhapsodieDataflowBundle';
public function getContainerExtension()
public function getContainerExtension(): ?ExtensionInterface
{
return new CodeRhapsodieDataflowExtension();
}
public function build(ContainerBuilder $container)
public function build(ContainerBuilder $container): void
{
$container->addCompilerPass(new DataflowTypeCompilerPass());
$container
->addCompilerPass(new DataflowTypeCompilerPass())
->addCompilerPass(new DefaultLoggerCompilerPass())
->addCompilerPass(new BusCompilerPass())
;
}
}

View File

@@ -5,50 +5,34 @@ 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\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;
use Symfony\Component\Validator\Validator\ValidatorInterface;
use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
/**
* @codeCoverageIgnore
*/
#[AsCommand('code-rhapsodie:dataflow:schedule:add', 'Create a scheduled dataflow')]
class AddScheduledDataflowCommand extends Command
{
protected static $defaultName = 'code-rhapsodie:dataflow:schedule:add';
/** @var DataflowTypeRegistryInterface */
private $registry;
/** @var ScheduledDataflowRepository */
private $scheduledDataflowRepository;
/** @var ValidatorInterface */
private $validator;
/** @var ConnectionFactory */
private $connectionFactory;
public function __construct(DataflowTypeRegistryInterface $registry, ScheduledDataflowRepository $scheduledDataflowRepository, ValidatorInterface $validator, ConnectionFactory $connectionFactory)
public function __construct(private DataflowTypeRegistryInterface $registry, private ScheduledDataflowRepository $scheduledDataflowRepository, private ValidatorInterface $validator, private ConnectionFactory $connectionFactory)
{
parent::__construct();
$this->registry = $registry;
$this->scheduledDataflowRepository = $scheduledDataflowRepository;
$this->validator = $validator;
$this->connectionFactory = $connectionFactory;
}
/**
* {@inheritdoc}
*/
protected function configure()
protected function configure(): void
{
$this
->setDescription('Create a scheduled dataflow')
->setHelp('The <info>%command.name%</info> allows you to create a new scheduled dataflow.')
->addOption('label', null, InputOption::VALUE_REQUIRED, 'Label of the scheduled dataflow')
->addOption('type', null, InputOption::VALUE_REQUIRED, 'Type of the scheduled dataflow (FQCN)')
@@ -63,7 +47,7 @@ class AddScheduledDataflowCommand extends Command
/**
* {@inheritdoc}
*/
protected function execute(InputInterface $input, OutputInterface $output)
protected function execute(InputInterface $input, OutputInterface $output): int
{
if (null !== $input->getOption('connection')) {
$this->connectionFactory->setConnectionName($input->getOption('connection'));
@@ -108,9 +92,9 @@ class AddScheduledDataflowCommand extends Command
'id' => null,
'label' => $label,
'dataflow_type' => $type,
'options' => json_decode($options, true),
'options' => json_decode($options, true, 512, JSON_THROW_ON_ERROR),
'frequency' => $frequency,
'next' => new \DateTimeImmutable($firstRun),
'next' => new \DateTime($firstRun),
'enabled' => $enabled,
]);

View File

@@ -5,43 +5,33 @@ 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\Attribute\AsCommand;
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;
use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
/**
* @codeCoverageIgnore
*/
#[AsCommand('code-rhapsodie:dataflow:schedule:change-status', 'Change schedule status')]
class ChangeScheduleStatusCommand extends Command
{
protected static $defaultName = 'code-rhapsodie:dataflow:schedule:change-status';
/** @var ScheduledDataflowRepository */
private $scheduledDataflowRepository;
/** @var ConnectionFactory */
private $connectionFactory;
public function __construct(ScheduledDataflowRepository $scheduledDataflowRepository, ConnectionFactory $connectionFactory)
public function __construct(private ScheduledDataflowRepository $scheduledDataflowRepository, private ConnectionFactory $connectionFactory)
{
parent::__construct();
$this->scheduledDataflowRepository = $scheduledDataflowRepository;
$this->connectionFactory = $connectionFactory;
}
/**
* {@inheritdoc}
*/
protected function configure()
protected function configure(): void
{
$this
->setDescription('Change schedule status')
->setHelp('The <info>%command.name%</info> command able you to change schedule status.')
->addArgument('schedule-id', InputArgument::REQUIRED, 'Id of the schedule')
->addOption('enable', null, InputOption::VALUE_NONE, 'Enable the schedule')
@@ -52,7 +42,7 @@ class ChangeScheduleStatusCommand extends Command
/**
* {@inheritdoc}
*/
protected function execute(InputInterface $input, OutputInterface $output)
protected function execute(InputInterface $input, OutputInterface $output): int
{
if (null !== $input->getOption('connection')) {
$this->connectionFactory->setConnectionName($input->getOption('connection'));

View File

@@ -0,0 +1,113 @@
<?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\Comparator;
use Doctrine\DBAL\Schema\Schema;
use Doctrine\DBAL\Schema\Table;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Question\ConfirmationQuestion;
use Symfony\Component\Console\Style\SymfonyStyle;
#[AsCommand(name: 'code-rhapsodie:dataflow:database-schema', description: 'Generates schema create / update SQL queries')]
class DatabaseSchemaCommand extends Command
{
public function __construct(private ConnectionFactory $connectionFactory)
{
parent::__construct();
}
/**
* {@inheritdoc}
*/
protected function configure(): void
{
$this
->setHelp('The <info>%command.name%</info> help you to generate SQL Query to create or update your database schema for this bundle')
->addOption('dump-sql', null, InputOption::VALUE_NONE, 'Dump only the update SQL queries.')
->addOption('update', null, InputOption::VALUE_NONE, 'Dump/execute only the update SQL queries.')
->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
}
protected function execute(InputInterface $input, OutputInterface $output): int
{
$io = new SymfonyStyle($input, $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->createSchemaManager();
$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->listSchemaNames();
}
$sequences = [];
if ($connection->getDatabasePlatform()->supportsSequences()) {
$sequences = $sm->listSequences();
}
$oldSchema = new Schema($tables, $sequences, $sm->createSchemaConfig(), $namespaces);
$sqls = $connection->getDatabasePlatform()->getAlterSchemaSQL((new Comparator($connection->getDatabasePlatform()))->compareSchemas($oldSchema, $schema));
if (empty($sqls)) {
$io->info('There is no update SQL queries.');
}
}
if ($input->getOption('dump-sql')) {
$io->text('Execute these SQL Queries on your database:');
foreach ($sqls as $sql) {
$io->text($sql . ';');
}
return Command::SUCCESS;
}
if (!$io->askQuestion(new ConfirmationQuestion('Are you sure to update database ?', true))) {
$io->text("Execution canceled.");
return Command::SUCCESS;
}
foreach ($sqls as $sql) {
$connection->executeQuery($sql);
}
$io->success(sprintf('%d queries executed.', \count($sqls)));
return parent::SUCCESS;
}
}

View File

@@ -4,44 +4,39 @@ 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\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
use Symfony\Component\Console\Style\SymfonyStyle;
/**
* Runs one dataflow.
*
* @codeCoverageIgnore
*/
class ExecuteDataflowCommand extends Command
#[AsCommand('code-rhapsodie:dataflow:execute', 'Runs one dataflow type with provided options')]
class ExecuteDataflowCommand extends Command implements LoggerAwareInterface
{
protected static $defaultName = 'code-rhapsodie:dataflow:execute';
use LoggerAwareTrait;
/** @var DataflowTypeRegistryInterface */
private $registry;
/** @var ConnectionFactory */
private $connectionFactory;
public function __construct(DataflowTypeRegistryInterface $registry, ConnectionFactory $connectionFactory)
public function __construct(private DataflowTypeRegistryInterface $registry, private ConnectionFactory $connectionFactory)
{
parent::__construct();
$this->registry = $registry;
$this->connectionFactory = $connectionFactory;
}
/**
* {@inheritdoc}
*/
protected function configure()
protected function configure(): void
{
$this
->setDescription('Runs one dataflow type with provided options')
->setHelp(<<<'EOF'
The <info>%command.name%</info> command runs one dataflow with the provided options.
@@ -56,21 +51,32 @@ EOF
/**
* {@inheritdoc}
*/
protected function execute(InputInterface $input, OutputInterface $output)
protected function execute(InputInterface $input, OutputInterface $output): int
{
if (null !== $input->getOption('connection')) {
$this->connectionFactory->setConnectionName($input->getOption('connection'));
}
$fqcnOrAlias = $input->getArgument('fqcn');
$options = json_decode($input->getArgument('options'), true);
$options = json_decode($input->getArgument('options'), true, 512, JSON_THROW_ON_ERROR);
$io = new SymfonyStyle($input, $output);
$dataflowType = $this->registry->getDataflowType($fqcnOrAlias);
if ($dataflowType instanceof LoggerAwareInterface && isset($this->logger)) {
$dataflowType->setLogger($this->logger);
}
$result = $dataflowType->process($options);
$output->writeln('Executed: '.$result->getName());
$output->writeln('Start time: '.$result->getStartTime()->format('Y/m/d H:i:s'));
$output->writeln('End time: '.$result->getEndTime()->format('Y/m/d H:i:s'));
$output->writeln('Success: '.$result->getSuccessCount());
$io->writeln('Executed: '.$result->getName());
$io->writeln('Start time: '.$result->getStartTime()->format('Y/m/d H:i:s'));
$io->writeln('End time: '.$result->getEndTime()->format('Y/m/d H:i:s'));
$io->writeln('Success: '.$result->getSuccessCount());
if ($result->hasErrors()) {
$io->error("Errors: {$result->getErrorCount()}\nExceptions traces are available in the logs.");
return 1;
}
return 0;
}

View File

@@ -5,17 +5,19 @@ 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\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;
use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
/**
* @codeCoverageIgnore
*/
#[AsCommand('code-rhapsodie:dataflow:job:show', 'Display job details for schedule or specific job')]
class JobShowCommand extends Command
{
private const STATUS_MAPPING = [
@@ -24,29 +26,17 @@ class JobShowCommand extends Command
Job::STATUS_COMPLETED => 'Completed',
];
protected static $defaultName = 'code-rhapsodie:dataflow:job:show';
/** @var JobRepository */
private $jobRepository;
/** @var ConnectionFactory */
private $connectionFactory;
public function __construct(JobRepository $jobRepository, ConnectionFactory $connectionFactory)
public function __construct(private JobRepository $jobRepository, private ConnectionFactory $connectionFactory)
{
parent::__construct();
$this->jobRepository = $jobRepository;
$this->connectionFactory = $connectionFactory;
}
/**
* {@inheritdoc}
*/
protected function configure()
protected function configure(): void
{
$this
->setDescription('Display job details for schedule or specific job')
->setHelp('The <info>%command.name%</info> display job details for schedule or specific job.')
->addOption('job-id', null, InputOption::VALUE_REQUIRED, 'Id of the job to get details')
->addOption('schedule-id', null, InputOption::VALUE_REQUIRED, 'Id of schedule for last execution details')
@@ -57,7 +47,7 @@ class JobShowCommand extends Command
/**
* {@inheritdoc}
*/
protected function execute(InputInterface $input, OutputInterface $output)
protected function execute(InputInterface $input, OutputInterface $output): int
{
if (null !== $input->getOption('connection')) {
$this->connectionFactory->setConnectionName($input->getOption('connection'));
@@ -97,21 +87,19 @@ class JobShowCommand extends Command
['Started at', $job->getStartTime() ? $job->getStartTime()->format('Y-m-d H:i:s') : '-'],
['Ended at', $job->getEndTime() ? $job->getEndTime()->format('Y-m-d H:i:s') : '-'],
['Object number', $job->getCount()],
['Errors', count($job->getExceptions())],
['Errors', count((array) $job->getExceptions())],
['Status', $this->translateStatus($job->getStatus())],
];
if ($input->getOption('details')) {
$display[] = ['Type', $job->getDataflowType()];
$display[] = ['Options', json_encode($job->getOptions())];
$display[] = ['Options', json_encode($job->getOptions(), JSON_THROW_ON_ERROR)];
$io->section('Summary');
}
$io->table(['Field', 'Value'], $display);
if ($input->getOption('details')) {
$io->section('Exceptions');
$exceptions = array_map(function (string $exception) {
return substr($exception, 0, 900).'…';
}, $job->getExceptions());
$exceptions = array_map(fn(string $exception) => substr($exception, 0, 900).'…', $job->getExceptions());
$io->write($exceptions);
}

View File

@@ -4,51 +4,37 @@ 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\Attribute\AsCommand;
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;
use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
/**
* Runs dataflows according to user-defined schedule.
*
* @codeCoverageIgnore
*/
#[AsCommand('code-rhapsodie:dataflow:run-pending', 'Runs dataflows based on the scheduled defined in the UI.')]
class RunPendingDataflowsCommand extends Command
{
use LockableTrait;
protected static $defaultName = 'code-rhapsodie:dataflow:run-pending';
/** @var ScheduledDataflowManagerInterface */
private $manager;
/** @var PendingDataflowRunnerInterface */
private $runner;
/** @var ConnectionFactory */
private $connectionFactory;
public function __construct(ScheduledDataflowManagerInterface $manager, PendingDataflowRunnerInterface $runner, ConnectionFactory $connectionFactory)
public function __construct(private ScheduledDataflowManagerInterface $manager, private PendingDataflowRunnerInterface $runner, private ConnectionFactory $connectionFactory)
{
parent::__construct();
$this->manager = $manager;
$this->runner = $runner;
$this->connectionFactory = $connectionFactory;
}
/**
* {@inheritdoc}
*/
protected function configure()
protected function configure(): void
{
$this
->setDescription('Runs dataflows based on the scheduled defined in the UI.')
->setHelp(<<<'EOF'
The <info>%command.name%</info> command runs dataflows according to the schedule defined in the UI by the user.
EOF
@@ -59,7 +45,7 @@ EOF
/**
* {@inheritdoc}
*/
protected function execute(InputInterface $input, OutputInterface $output)
protected function execute(InputInterface $input, OutputInterface $output): int
{
if (!$this->lock()) {
$output->writeln('The command is already running in another process.');

View File

@@ -4,42 +4,32 @@ declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\Command;
use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;
use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
/**
* @codeCoverageIgnore
*/
#[AsCommand('code-rhapsodie:dataflow:schedule:list', 'List scheduled dataflows')]
class ScheduleListCommand extends Command
{
protected static $defaultName = 'code-rhapsodie:dataflow:schedule:list';
/** @var ScheduledDataflowRepository */
private $scheduledDataflowRepository;
/** @var ConnectionFactory */
private $connectionFactory;
public function __construct(ScheduledDataflowRepository $scheduledDataflowRepository, ConnectionFactory $connectionFactory)
public function __construct(private ScheduledDataflowRepository $scheduledDataflowRepository, private ConnectionFactory $connectionFactory)
{
parent::__construct();
$this->scheduledDataflowRepository = $scheduledDataflowRepository;
$this->connectionFactory = $connectionFactory;
}
/**
* {@inheritdoc}
*/
protected function configure()
protected function configure(): void
{
$this
->setDescription('List scheduled dataflows')
->setHelp('The <info>%command.name%</info> lists all scheduled dataflows.')
->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
}
@@ -47,7 +37,7 @@ class ScheduleListCommand extends Command
/**
* {@inheritdoc}
*/
protected function execute(InputInterface $input, OutputInterface $output)
protected function execute(InputInterface $input, OutputInterface $output): int
{
if (null !== $input->getOption('connection')) {
$this->connectionFactory->setConnectionName($input->getOption('connection'));

View File

@@ -4,42 +4,34 @@ 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\Comparator;
use Doctrine\DBAL\Schema\Schema;
use Doctrine\DBAL\Schema\Table;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\ArrayInput;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;
use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
/**
* @codeCoverageIgnore
* @deprecated This command is deprecated and will be removed in 6.0, use this command "code-rhapsodie:dataflow:database-schema" instead.
*/
#[AsCommand('code-rhapsodie:dataflow:dump-schema', 'Generates schema create / update SQL queries')]
class SchemaCommand extends Command
{
protected static $defaultName = 'code-rhapsodie:dataflow:dump-schema';
/** @var ConnectionFactory */
private $connectionFactory;
public function __construct(ConnectionFactory $connectionFactory)
{
parent::__construct();
$this->connectionFactory = $connectionFactory;
}
/**
* {@inheritdoc}
*/
protected function configure()
protected function configure(): void
{
$this
->setDescription('Generates schema create / update SQL queries')
->setHelp('The <info>%command.name%</info> help you to generate SQL Query to create or update your database schema for this bundle')
->addOption('update', null, InputOption::VALUE_NONE, 'Dump only the update SQL queries.')
->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use')
@@ -49,51 +41,26 @@ class SchemaCommand extends Command
/**
* {@inheritdoc}
*/
protected function execute(InputInterface $input, OutputInterface $output)
protected function execute(InputInterface $input, OutputInterface $output): int
{
if (null !== $input->getOption('connection')) {
$this->connectionFactory->setConnectionName($input->getOption('connection'));
}
$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);
}
$io->warning('This command is deprecated and will be removed in 6.0, use this command "code-rhapsodie:dataflow:database-schema" instead.');
$options = array_filter($input->getOptions());
//add -- before each keys
$options = array_combine(
array_map(fn($key) => '--' . $key, array_keys($options)),
array_values($options)
);
$options['--dump-sql'] = true;
$inputArray = new ArrayInput([
'command' => 'code-rhapsodie:dataflow:database-schema',
...$options
]);
return $this->getApplication()->doRun($inputArray, $output);
}
}

View File

@@ -0,0 +1,70 @@
<?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
{
private ?string $name = null;
private ?iterable $reader = null;
private array $steps = [];
/** @var WriterInterface[] */
private array $writers = [];
public function __construct(protected ?int $loopInterval = 0, protected ?int $emitInterval = 0)
{
}
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,10 +4,15 @@ declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\DataflowType;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerAwareTrait;
use Psr\Log\LoggerInterface;
use Symfony\Component\OptionsResolver\OptionsResolver;
abstract class AbstractDataflowType implements DataflowTypeInterface
abstract class AbstractDataflowType implements DataflowTypeInterface, LoggerAwareInterface
{
use LoggerAwareTrait;
/**
* @codeCoverageIgnore
*/
@@ -22,15 +27,22 @@ abstract class AbstractDataflowType implements DataflowTypeInterface
$this->configureOptions($optionsResolver);
$options = $optionsResolver->resolve($options);
$builder = (new DataflowBuilder())
->setName($this->getLabel())
;
$builder = $this->createDataflowBuilder();
$builder->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

@@ -0,0 +1,164 @@
<?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 callable[] */
private array $steps = [];
/** @var WriterInterface[] */
private array $writers = [];
private array $states = [];
private array $stepsJobs = [];
public function __construct(private iterable $reader, private ?string $name, private ?int $loopInterval = 0, private ?int $emitInterval = 0)
{
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 \DateTime();
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 \DateTime(), $count, $exceptions);
}
/**
* @param int $count internal count reference
* @param array $exceptions internal exceptions
*/
private function processState(mixed $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 ((is_countable($this->stepsJobs[$stepIndex]) ? count($this->stepsJobs[$stepIndex]) : 0) < $scale && !isset($this->stepsJobs[$stepIndex][$readIndex])) {
$this->stepsJobs[$stepIndex][$readIndex] = true;
/** @var Promise<void> $promise */
$promise = coroutine($step)($item);
$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,34 +6,26 @@ namespace CodeRhapsodie\DataflowBundle\DataflowType\Dataflow;
use CodeRhapsodie\DataflowBundle\DataflowType\Result;
use CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerAwareTrait;
class Dataflow implements DataflowInterface
class Dataflow implements DataflowInterface, LoggerAwareInterface
{
/** @var string */
private $name;
/** @var iterable */
private $reader;
use LoggerAwareTrait;
/** @var callable[] */
private $steps = [];
private array $steps = [];
/** @var WriterInterface[] */
private $writers = [];
private array $writers = [];
/**
* @param iterable $reader
* @param string|null $name
*/
public function __construct(iterable $reader, ?string $name)
private ?\Closure $customExceptionIndex = null;
public function __construct(private iterable $reader, private ?string $name)
{
$this->reader = $reader;
$this->name = $name;
}
/**
* @param callable $step
*
* @return $this
*/
public function addStep(callable $step): self
@@ -44,8 +36,6 @@ class Dataflow implements DataflowInterface
}
/**
* @param WriterInterface $writer
*
* @return $this
*/
public function addWriter(WriterInterface $writer): self
@@ -55,6 +45,16 @@ class Dataflow implements DataflowInterface
return $this;
}
/**
* @return $this
*/
public function setCustomExceptionIndex(callable $callable): self
{
$this->customExceptionIndex = \Closure::fromCallable($callable);
return $this;
}
/**
* {@inheritdoc}
*/
@@ -64,31 +64,43 @@ class Dataflow implements DataflowInterface
$exceptions = [];
$startTime = new \DateTime();
foreach ($this->writers as $writer) {
$writer->prepare();
}
foreach ($this->reader as $index => $item) {
try {
$this->processItem($item);
} catch (\Exception $e) {
$exceptions[$index] = $e;
try {
foreach ($this->writers as $writer) {
$writer->prepare();
}
++$count;
}
foreach ($this->reader as $index => $item) {
try {
$this->processItem($item);
} catch (\Throwable $e) {
$exceptionIndex = $index;
try {
if (is_callable($this->customExceptionIndex)) {
$exceptionIndex = (string) ($this->customExceptionIndex)($item, $index);
}
} catch (\Throwable $e2) {
$exceptions[$index] = $e2;
$this->logException($e2, $index);
}
$exceptions[$exceptionIndex] = $e;
$this->logException($e, $exceptionIndex);
}
foreach ($this->writers as $writer) {
$writer->finish();
++$count;
}
foreach ($this->writers as $writer) {
$writer->finish();
}
} catch (\Throwable $e) {
$exceptions[] = $e;
$this->logException($e);
}
return new Result($this->name, $startTime, new \DateTime(), $count, $exceptions);
}
/**
* @param mixed $item
*/
private function processItem($item): void
private function processItem(mixed $item): void
{
foreach ($this->steps as $step) {
$item = call_user_func($step, $item);
@@ -102,4 +114,13 @@ class Dataflow implements DataflowInterface
$writer->write($item);
}
}
private function logException(\Throwable $e, string|int|null $index = null): void
{
if (!isset($this->logger)) {
return;
}
$this->logger->error($e, ['exception' => $e, 'index' => $index]);
}
}

View File

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

View File

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

View File

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

@@ -0,0 +1,58 @@
<?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
{
/**
* CollectionWriter constructor.
*/
public function __construct(private WriterInterface $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.', get_debug_type($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

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

View File

@@ -0,0 +1,95 @@
<?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 array $delegates = [];
/**
* DelegatorWriter constructor.
*/
public function __construct()
{
}
/**
* {@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', get_debug_type($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

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

View File

@@ -4,11 +4,23 @@ 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();
public function write($item);
/**
* Write an item.
*/
public function write(mixed $item);
/**
* Called after the dataflow is processed.
*/
public function finish();
}

View File

@@ -28,5 +28,11 @@ class CodeRhapsodieDataflowExtension extends Extension
$config = $this->processConfiguration($configuration, $configs);
$container->setParameter('coderhapsodie.dataflow.dbal_default_connection', $config['dbal_default_connection']);
$container->setParameter('coderhapsodie.dataflow.default_logger', $config['default_logger']);
if ($config['messenger_mode']['enabled']) {
$container->setParameter('coderhapsodie.dataflow.bus', $config['messenger_mode']['bus']);
$loader->load('messenger_services.yaml');
}
}
}

View File

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

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

@@ -6,19 +6,38 @@ 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()
public function getConfigTreeBuilder(): \Symfony\Component\Config\Definition\Builder\TreeBuilder
{
$treeBuilder = new TreeBuilder();
$rootNode = $treeBuilder->root('code_rhapsodie_dataflow');
$treeBuilder = new TreeBuilder('code_rhapsodie_dataflow');
$rootNode = $treeBuilder->getRootNode();
$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 fn($v): bool => $v['enabled'] && !interface_exists(MessageBusInterface::class))
->thenInvalid('You need "symfony/messenger" in order to use Dataflow messenger mode.')
->end()
->end()
->end()
;

View File

@@ -13,9 +13,10 @@ use Symfony\Component\Validator\Constraints as Asserts;
*/
class Job
{
const STATUS_PENDING = 0;
const STATUS_RUNNING = 1;
const STATUS_COMPLETED = 2;
public const STATUS_PENDING = 0;
public const STATUS_RUNNING = 1;
public const STATUS_COMPLETED = 2;
public const STATUS_QUEUED = 3;
private const KEYS = [
'id',
@@ -31,78 +32,35 @@ class Job
'end_time',
];
/**
* @var int|null
*/
private $id;
private ?int $id = null;
/**
* @var int
*
* @Asserts\Range(min=0, max=2)
*/
private $status;
#[Asserts\Range(min: 0, max: 2)]
private int $status = self::STATUS_PENDING;
/**
* @var string|null
*
* @Asserts\NotBlank()
* @Asserts\Length(min=1, max=255)
* @Asserts\Regex("#^[[:alnum:] ]+\z#u")
*/
private $label;
#[Asserts\NotBlank]
#[Asserts\Length(min: 1, max: 255)]
#[Asserts\Regex('#^[[:alnum:] ]+\z#u')]
private ?string $label = null;
/**
* @var string|null
*
* @Asserts\NotBlank()
* @Asserts\Length(min=1, max=255)
* @Asserts\Regex("#^[[:alnum:]\\]+\z#u")
*/
private $dataflowType;
#[Asserts\NotBlank]
#[Asserts\Length(min: 1, max: 255)]
#[Asserts\Regex('#^[[:alnum:]\\\]+\z#u')]
private ?string $dataflowType = null;
/**
* @var array|null
*/
private $options;
private ?array $options = null;
/**
* @var \DateTimeInterface|null
*
* @Asserts\DateTime()
*/
private $requestedDate;
private ?\DateTimeInterface $requestedDate = null;
/**
* @var int|null
*/
private $scheduledDataflowId;
private ?int $scheduledDataflowId = null;
/**
* @var int|null
*/
private $count;
private ?int $count = 0;
/**
* @var array|null
*/
private $exceptions;
private ?array $exceptions = null;
/**
* @var \DateTimeInterface|null
*/
private $startTime;
private ?\DateTimeInterface $startTime = null;
/**
* @var \DateTimeInterface|null
*/
private $endTime;
private ?\DateTimeInterface $endTime = null;
/**
* @param ScheduledDataflow $scheduled
*
* @return Job
*/
public static function createFromScheduledDataflow(ScheduledDataflow $scheduled): self
{
return (new static())
@@ -114,18 +72,11 @@ class Job
->setScheduledDataflowId($scheduled->getId());
}
public function __construct()
{
$this->count = 0;
$this->status = static::STATUS_PENDING;
}
public static function createFromArray(array $datas)
{
$lost = array_diff(static::KEYS, array_keys($datas));
if (count($lost) > 0) {
throw new \LogicException('The first argument of '.__METHOD__.' must be contains: "'.implode(', ',
$lost).'"');
throw new \LogicException('The first argument of '.__METHOD__.' must be contains: "'.implode(', ', $lost).'"');
}
$job = new self();
@@ -161,11 +112,6 @@ class Job
];
}
/**
* @param int $id
*
* @return Job
*/
public function setId(int $id): Job
{
$this->id = $id;
@@ -173,27 +119,16 @@ class Job
return $this;
}
/**
* @return int|null
*/
public function getId(): ?int
{
return $this->id;
}
/**
* @return int
*/
public function getStatus(): int
{
return $this->status;
}
/**
* @param int $status
*
* @return Job
*/
public function setStatus(int $status): Job
{
$this->status = $status;
@@ -201,19 +136,11 @@ class Job
return $this;
}
/**
* @return string|null
*/
public function getLabel(): ?string
{
return $this->label;
}
/**
* @param string|null $label
*
* @return Job
*/
public function setLabel(?string $label): Job
{
$this->label = $label;
@@ -221,19 +148,11 @@ class Job
return $this;
}
/**
* @return string|null
*/
public function getDataflowType(): ?string
{
return $this->dataflowType;
}
/**
* @param string|null $dataflowType
*
* @return Job
*/
public function setDataflowType(?string $dataflowType): Job
{
$this->dataflowType = $dataflowType;
@@ -241,19 +160,11 @@ class Job
return $this;
}
/**
* @return array|null
*/
public function getOptions(): ?array
{
return $this->options;
}
/**
* @param array|null $options
*
* @return Job
*/
public function setOptions(?array $options): Job
{
$this->options = $options;
@@ -261,19 +172,11 @@ class Job
return $this;
}
/**
* @return \DateTimeInterface|null
*/
public function getRequestedDate(): ?\DateTimeInterface
{
return $this->requestedDate;
}
/**
* @param \DateTimeInterface|null $requestedDate
*
* @return Job
*/
public function setRequestedDate(?\DateTimeInterface $requestedDate): Job
{
$this->requestedDate = $requestedDate;
@@ -281,19 +184,11 @@ class Job
return $this;
}
/**
* @return int|null
*/
public function getScheduledDataflowId(): ?int
{
return $this->scheduledDataflowId;
}
/**
* @param int|null $scheduledDataflowId
*
* @return Job
*/
public function setScheduledDataflowId(?int $scheduledDataflowId): Job
{
$this->scheduledDataflowId = $scheduledDataflowId;
@@ -301,19 +196,11 @@ class Job
return $this;
}
/**
* @return int|null
*/
public function getCount(): ?int
{
return $this->count;
}
/**
* @param int|null $count
*
* @return Job
*/
public function setCount(?int $count): Job
{
$this->count = $count;
@@ -321,19 +208,11 @@ class Job
return $this;
}
/**
* @return array|null
*/
public function getExceptions(): ?array
{
return $this->exceptions;
}
/**
* @param array|null $exceptions
*
* @return Job
*/
public function setExceptions(?array $exceptions): Job
{
$this->exceptions = $exceptions;
@@ -341,19 +220,11 @@ class Job
return $this;
}
/**
* @return \DateTimeInterface|null
*/
public function getStartTime(): ?\DateTimeInterface
{
return $this->startTime;
}
/**
* @param \DateTimeInterface|null $startTime
*
* @return Job
*/
public function setStartTime(?\DateTimeInterface $startTime): Job
{
$this->startTime = $startTime;
@@ -361,19 +232,11 @@ class Job
return $this;
}
/**
* @return \DateTimeInterface|null
*/
public function getEndTime(): ?\DateTimeInterface
{
return $this->endTime;
}
/**
* @param \DateTimeInterface|null $endTime
*
* @return Job
*/
public function setEndTime(?\DateTimeInterface $endTime): Job
{
$this->endTime = $endTime;

View File

@@ -14,7 +14,7 @@ use Symfony\Component\Validator\Constraints as Asserts;
*/
class ScheduledDataflow
{
const AVAILABLE_FREQUENCIES = [
public const AVAILABLE_FREQUENCIES = [
'1 hour',
'1 day',
'1 week',
@@ -23,58 +23,35 @@ class ScheduledDataflow
private const KEYS = ['id', 'label', 'dataflow_type', 'options', 'frequency', 'next', 'enabled'];
/**
* @var int|null
*/
private $id;
private ?int $id = null;
#[Asserts\NotBlank]
#[Asserts\Length(min: 1, max: 255)]
#[Asserts\Regex('#^[[:alnum:] ]+\z#u')]
private ?string $label = null;
#[Asserts\NotBlank]
#[Asserts\Length(min: 1, max: 255)]
#[Asserts\Regex('#^[[:alnum:]\\\]+\z#u')]
private ?string $dataflowType = null;
private ?array $options = null;
/**
* @var string|null
*
* @Asserts\NotBlank()
* @Asserts\Length(min=1, max=255)
* @Asserts\Regex("#^[[:alnum:] ]+\z#u")
*/
private $label;
/**
* @var string|null
*
* @Asserts\NotBlank()
* @Asserts\Length(min=1, max=255)
* @Asserts\Regex("#^[[:alnum:]\\]+\z#u")
*/
private $dataflowType;
/**
* @var array|null
*/
private $options;
/**
* @var string|null
*
* @Asserts\NotBlank()
* @Frequency()
*/
private $frequency;
#[Asserts\NotBlank]
private ?string $frequency = null;
/**
* @var \DateTimeInterface|null
*/
private $next;
private ?\DateTimeInterface $next = null;
/**
* @var bool|null
*/
private $enabled;
private ?bool $enabled = null;
public static function createFromArray(array $datas)
{
$lost = array_diff(static::KEYS, array_keys($datas));
if (count($lost) > 0) {
throw new \LogicException('The first argument of '.__METHOD__.' must be contains: "'.implode(', ',
$lost).'"');
throw new \LogicException('The first argument of '.__METHOD__.' must be contains: "'.implode(', ', $lost).'"');
}
$scheduledDataflow = new self();
@@ -103,11 +80,6 @@ class ScheduledDataflow
];
}
/**
* @param int $id
*
* @return ScheduledDataflow
*/
public function setId(int $id): ScheduledDataflow
{
$this->id = $id;
@@ -115,27 +87,16 @@ class ScheduledDataflow
return $this;
}
/**
* @return int|null
*/
public function getId(): ?int
{
return $this->id;
}
/**
* @return string|null
*/
public function getLabel(): ?string
{
return $this->label;
}
/**
* @param string|null $label
*
* @return ScheduledDataflow
*/
public function setLabel(?string $label): ScheduledDataflow
{
$this->label = $label;
@@ -143,19 +104,11 @@ class ScheduledDataflow
return $this;
}
/**
* @return string|null
*/
public function getDataflowType(): ?string
{
return $this->dataflowType;
}
/**
* @param string|null $dataflowType
*
* @return ScheduledDataflow
*/
public function setDataflowType(?string $dataflowType): ScheduledDataflow
{
$this->dataflowType = $dataflowType;
@@ -163,19 +116,11 @@ class ScheduledDataflow
return $this;
}
/**
* @return array|null
*/
public function getOptions(): ?array
{
return $this->options;
}
/**
* @param array|null $options
*
* @return ScheduledDataflow
*/
public function setOptions(?array $options): ScheduledDataflow
{
$this->options = $options;
@@ -183,19 +128,11 @@ class ScheduledDataflow
return $this;
}
/**
* @return string|null
*/
public function getFrequency(): ?string
{
return $this->frequency;
}
/**
* @param string|null $frequency
*
* @return ScheduledDataflow
*/
public function setFrequency(?string $frequency): ScheduledDataflow
{
$this->frequency = $frequency;
@@ -203,19 +140,11 @@ class ScheduledDataflow
return $this;
}
/**
* @return \DateTimeInterface|null
*/
public function getNext(): ?\DateTimeInterface
{
return $this->next;
}
/**
* @param \DateTimeInterface|null $next
*
* @return ScheduledDataflow
*/
public function setNext(?\DateTimeInterface $next): ScheduledDataflow
{
$this->next = $next;
@@ -223,19 +152,11 @@ class ScheduledDataflow
return $this;
}
/**
* @return bool|null
*/
public function getEnabled(): ?bool
{
return $this->enabled;
}
/**
* @param bool|null $enabled
*
* @return ScheduledDataflow
*/
public function setEnabled(?bool $enabled): ScheduledDataflow
{
$this->enabled = $enabled;

14
src/Event/CrEvent.php Normal file
View File

@@ -0,0 +1,14 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\Event;
use Symfony\Contracts\EventDispatcher\Event;
/**
* @codeCoverageIgnore
*/
abstract class CrEvent extends Event
{
}

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -0,0 +1,36 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\Logger;
use Monolog\Formatter\FormatterInterface;
use Monolog\Formatter\LineFormatter;
use Monolog\Handler\AbstractProcessingHandler;
use Monolog\Logger;
use Monolog\LogRecord;
class BufferHandler extends AbstractProcessingHandler
{
private const FORMAT = "[%datetime%] %level_name% when processing item %context.index%: %message% %context% %extra%\n";
private array $buffer = [];
public function clearBuffer(): array
{
$logs = $this->buffer;
$this->buffer = [];
return $logs;
}
protected function write(array|LogRecord $record): void
{
$this->buffer[] = $record['formatted'];
}
protected function getDefaultFormatter(): FormatterInterface
{
return new LineFormatter(self::FORMAT);
}
}

View File

@@ -0,0 +1,32 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\Logger;
use Psr\Log\AbstractLogger;
use Psr\Log\LoggerInterface;
final class DelegatingLogger extends AbstractLogger
{
/** @var LoggerInterface[] */
private ?array $loggers = null;
public function __construct(iterable $loggers)
{
foreach ($loggers as $logger) {
if (!$logger instanceof LoggerInterface) {
throw new \InvalidArgumentException(sprintf('Only instances of %s should be passed to the constructor of %s. An instance of %s was passed instead.', LoggerInterface::class, self::class, $logger::class));
}
$this->loggers[] = $logger;
}
}
public function log($level, $message, array $context = []): void
{
foreach ($this->loggers as $logger) {
$logger->log($level, $message, $context);
}
}
}

View File

@@ -4,31 +4,19 @@ declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\Manager;
use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
use CodeRhapsodie\DataflowBundle\Entity\Job;
use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
use Doctrine\DBAL\Driver\Connection;
use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
use Doctrine\DBAL\Connection;
/**
* Handles scheduled dataflows execution dates based on their frequency.
*/
class ScheduledDataflowManager implements ScheduledDataflowManagerInterface
{
/** @var ScheduledDataflowRepository */
private $scheduledDataflowRepository;
/** @var JobRepository */
private $jobRepository;
/** @var Connection */
private $connection;
public function __construct(Connection $connection, ScheduledDataflowRepository $scheduledDataflowRepository, JobRepository $jobRepository)
public function __construct(private Connection $connection, private ScheduledDataflowRepository $scheduledDataflowRepository, private JobRepository $jobRepository)
{
$this->connection = $connection;
$this->scheduledDataflowRepository = $scheduledDataflowRepository;
$this->jobRepository = $jobRepository;
}
/**
@@ -53,9 +41,6 @@ class ScheduledDataflowManager implements ScheduledDataflowManagerInterface
$this->connection->commit();
}
/**
* @param ScheduledDataflow $scheduled
*/
private function updateScheduledDataflowNext(ScheduledDataflow $scheduled): void
{
$interval = \DateInterval::createFromDateString($scheduled->getFrequency());
@@ -70,9 +55,6 @@ class ScheduledDataflowManager implements ScheduledDataflowManagerInterface
$this->scheduledDataflowRepository->save($scheduled);
}
/**
* @param ScheduledDataflow $scheduled
*/
private function createPendingForScheduled(ScheduledDataflow $scheduled): void
{
$this->jobRepository->save(Job::createFromScheduledDataflow($scheduled));

View File

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

View File

@@ -0,0 +1,22 @@
<?php
declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\MessengerMode;
use CodeRhapsodie\DataflowBundle\Processor\JobProcessorInterface;
use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
#[AsMessageHandler]
class JobMessageHandler
{
public function __construct(private JobRepository $repository, private JobProcessorInterface $processor)
{
}
public function __invoke(JobMessage $message)
{
$this->processor->process($this->repository->find($message->getJobId()));
}
}

View File

@@ -0,0 +1,77 @@
<?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;
public function __construct(private JobRepository $repository, private DataflowTypeRegistryInterface $registry, private EventDispatcherInterface $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
{
$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);
$this->dispatcher->dispatch(new ProcessingEvent($job), Events::AFTER_PROCESSING);
}
}

View File

@@ -0,0 +1,12 @@
<?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\Exceptions\UnknownDataflowTypeException;
use CodeRhapsodie\DataflowBundle\DataflowType\DataflowTypeInterface;
use CodeRhapsodie\DataflowBundle\Exceptions\UnknownDataflowTypeException;
/**
* Array based dataflow types registry.
@@ -13,10 +13,10 @@ use CodeRhapsodie\DataflowBundle\DataflowType\DataflowTypeInterface;
class DataflowTypeRegistry implements DataflowTypeRegistryInterface
{
/** @var array|DataflowTypeInterface[] */
private $fqcnRegistry = [];
private array $fqcnRegistry = [];
/** @var array|DataflowTypeInterface[] */
private $aliasesRegistry = [];
private array $aliasesRegistry = [];
/**
* {@inheritdoc}
@@ -31,7 +31,7 @@ class DataflowTypeRegistry implements DataflowTypeRegistryInterface
return $this->aliasesRegistry[$fqcnOrAlias];
}
throw new UnknownDataflowTypeException($fqcnOrAlias);
throw UnknownDataflowTypeException::create($fqcnOrAlias, [...array_keys($this->fqcnRegistry), ...array_keys($this->aliasesRegistry)]);
}
/**
@@ -47,7 +47,7 @@ class DataflowTypeRegistry implements DataflowTypeRegistryInterface
*/
public function registerDataflowType(DataflowTypeInterface $dataflowType): void
{
$this->fqcnRegistry[get_class($dataflowType)] = $dataflowType;
$this->fqcnRegistry[$dataflowType::class] = $dataflowType;
foreach ($dataflowType->getAliases() as $alias) {
$this->aliasesRegistry[$alias] = $dataflowType;
}

View File

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

View File

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

View File

@@ -6,7 +6,8 @@ namespace CodeRhapsodie\DataflowBundle\Repository;
use CodeRhapsodie\DataflowBundle\Entity\Job;
use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
use Doctrine\DBAL\Driver\Connection;
use Doctrine\DBAL\Connection;
use Doctrine\DBAL\ParameterType;
use Doctrine\DBAL\Query\QueryBuilder;
/**
@@ -20,40 +21,18 @@ class JobRepository
public const TABLE_NAME = 'cr_dataflow_job';
private const FIELDS_TYPE = [
'id' => \PDO::PARAM_INT,
'status' => \PDO::PARAM_INT,
'label' => \PDO::PARAM_STR,
'dataflow_type' => \PDO::PARAM_STR,
'options' => \PDO::PARAM_STR,
'requested_date' => 'datetime',
'scheduled_dataflow_id' => \PDO::PARAM_INT,
'count' => \PDO::PARAM_INT,
'exceptions' => \PDO::PARAM_STR,
'start_time' => 'datetime',
'end_time' => 'datetime',
];
/**
* @var \Doctrine\DBAL\Connection
*/
private $connection;
/**
* JobRepository constructor.
*
* @param Connection $connection
*/
public function __construct(Connection $connection)
public function __construct(private Connection $connection)
{
$this->connection = $connection;
}
public function find(int $jobId)
{
$qb = $this->createQueryBuilder();
$qb
->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($jobId, \PDO::PARAM_INT)))
->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($jobId, ParameterType::INTEGER)))
;
return $this->returnFirstOrNull($qb);
@@ -64,12 +43,12 @@ class JobRepository
$qb = $this->createQueryBuilder();
$qb
->andWhere($qb->expr()->isNull('scheduled_dataflow_id'))
->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, \PDO::PARAM_INT)));
$stmt = $qb->execute();
->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, ParameterType::INTEGER)));
$stmt = $qb->executeQuery();
if (0 === $stmt->rowCount()) {
return [];
}
while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
while (false !== ($row = $stmt->fetchAssociative())) {
yield Job::createFromArray($this->initDateTime($this->initArray($row)));
}
}
@@ -78,8 +57,8 @@ class JobRepository
{
$qb = $this->createQueryBuilder();
$qb
->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($scheduled->getId(), \PDO::PARAM_INT)))
->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, \PDO::PARAM_INT)));
->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($scheduled->getId(), ParameterType::INTEGER)))
->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, ParameterType::INTEGER)));
return $this->returnFirstOrNull($qb);
}
@@ -88,7 +67,7 @@ class JobRepository
{
$qb = $this->createQueryBuilder();
$qb->andWhere($qb->expr()->lte('requested_date', $qb->createNamedParameter(new \DateTime(), 'datetime')))
->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, \PDO::PARAM_INT)))
->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, ParameterType::INTEGER)))
->orderBy('requested_date', 'ASC')
->setMaxResults(1)
;
@@ -99,7 +78,7 @@ class JobRepository
public function findLastForDataflowId(int $dataflowId): ?Job
{
$qb = $this->createQueryBuilder();
$qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($dataflowId, \PDO::PARAM_INT)))
$qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($dataflowId, ParameterType::INTEGER)))
->orderBy('requested_date', 'DESC')
->setMaxResults(1)
;
@@ -113,11 +92,11 @@ class JobRepository
$qb
->orderBy('requested_date', 'DESC')
->setMaxResults(20);
$stmt = $qb->execute();
$stmt = $qb->executeQuery();
if (0 === $stmt->rowCount()) {
return [];
}
while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
while (false !== ($row = $stmt->fetchAssociative())) {
yield Job::createFromArray($row);
}
}
@@ -125,14 +104,14 @@ class JobRepository
public function findForScheduled(int $id): iterable
{
$qb = $this->createQueryBuilder();
$qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($id, \PDO::PARAM_INT)))
$qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($id, ParameterType::INTEGER)))
->orderBy('requested_date', 'DESC')
->setMaxResults(20);
$stmt = $qb->execute();
$stmt = $qb->executeQuery();
if (0 === $stmt->rowCount()) {
return [];
}
while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
while (false !== ($row = $stmt->fetchAssociative())) {
yield Job::createFromArray($row);
}
}
@@ -143,19 +122,19 @@ class JobRepository
unset($datas['id']);
if (is_array($datas['options'])) {
$datas['options'] = json_encode($datas['options']);
$datas['options'] = json_encode($datas['options'], JSON_THROW_ON_ERROR);
}
if (is_array($datas['exceptions'])) {
$datas['exceptions'] = json_encode($datas['exceptions']);
$datas['exceptions'] = json_encode($datas['exceptions'], JSON_THROW_ON_ERROR);
}
if (null === $job->getId()) {
$this->connection->insert(static::TABLE_NAME, $datas, static::FIELDS_TYPE);
$this->connection->insert(static::TABLE_NAME, $datas, $this->getFields());
$job->setId((int) $this->connection->lastInsertId());
return;
}
$this->connection->update(static::TABLE_NAME, $datas, ['id' => $job->getId()], static::FIELDS_TYPE);
$this->connection->update(static::TABLE_NAME, $datas, ['id' => $job->getId()], $this->getFields());
}
public function createQueryBuilder($alias = null): QueryBuilder
@@ -169,11 +148,28 @@ class JobRepository
private function returnFirstOrNull(QueryBuilder $qb): ?Job
{
$stmt = $qb->execute();
$stmt = $qb->executeQuery();
if (0 === $stmt->rowCount()) {
return null;
}
return Job::createFromArray($this->initDateTime($this->initArray($stmt->fetch(\PDO::FETCH_ASSOC))));
return Job::createFromArray($this->initDateTime($this->initArray($stmt->fetchAssociative())));
}
private function getFields(): array
{
return [
'id' => ParameterType::INTEGER,
'status' => ParameterType::INTEGER,
'label' => ParameterType::STRING,
'dataflow_type' => ParameterType::STRING,
'options' => ParameterType::STRING,
'requested_date' => 'datetime',
'scheduled_dataflow_id' => ParameterType::INTEGER,
'count' => ParameterType::INTEGER,
'exceptions' => ParameterType::STRING,
'start_time' => 'datetime',
'end_time' => 'datetime',
];
}
}

View File

@@ -5,7 +5,8 @@ declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\Repository;
use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
use Doctrine\DBAL\Driver\Connection;
use Doctrine\DBAL\Connection;
use Doctrine\DBAL\ParameterType;
use Doctrine\DBAL\Query\QueryBuilder;
/**
@@ -19,28 +20,11 @@ class ScheduledDataflowRepository
public const TABLE_NAME = 'cr_dataflow_scheduled';
private const FIELDS_TYPE = [
'id' => \PDO::PARAM_INT,
'label' => \PDO::PARAM_STR,
'dataflow_type' => \PDO::PARAM_STR,
'options' => \PDO::PARAM_STR,
'frequency' => \PDO::PARAM_STR,
'next' => 'datetime',
'enabled' => \PDO::PARAM_BOOL,
];
/**
* @var \Doctrine\DBAL\Connection
*/
private $connection;
/**
* JobRepository constructor.
*
* @param Connection $connection
*/
public function __construct(Connection $connection)
public function __construct(private Connection $connection)
{
$this->connection = $connection;
}
/**
@@ -52,15 +36,15 @@ class ScheduledDataflowRepository
{
$qb = $this->createQueryBuilder();
$qb->andWhere($qb->expr()->lte('next', $qb->createNamedParameter(new \DateTime(), 'datetime')))
->andWhere($qb->expr()->eq('enabled', 1))
->andWhere($qb->expr()->eq('enabled', $qb->createNamedParameter(1, ParameterType::INTEGER)))
->orderBy('next', 'ASC')
;
;
$stmt = $qb->execute();
$stmt = $qb->executeQuery();
if (0 === $stmt->rowCount()) {
return [];
}
while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
while (false !== ($row = $stmt->fetchAssociative())) {
yield ScheduledDataflow::createFromArray($this->initDateTime($this->initArray($row)));
}
}
@@ -68,7 +52,7 @@ class ScheduledDataflowRepository
public function find(int $scheduleId): ?ScheduledDataflow
{
$qb = $this->createQueryBuilder();
$qb->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($scheduleId, \PDO::PARAM_INT)))
$qb->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($scheduleId, ParameterType::INTEGER)))
->setMaxResults(1)
;
@@ -80,12 +64,12 @@ class ScheduledDataflowRepository
$qb = $this->createQueryBuilder();
$qb->orderBy('label', 'ASC');
$stmt = $qb->execute();
$stmt = $qb->executeQuery();
if (0 === $stmt->rowCount()) {
return [];
}
while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
yield ScheduledDataflow::createFromArray($this->initDateTime($this->initOptions($row)));
while (false !== ($row = $stmt->fetchAssociative())) {
yield ScheduledDataflow::createFromArray($this->initDateTime($this->initArray($row)));
}
}
@@ -98,7 +82,7 @@ class ScheduledDataflowRepository
->orderBy('w.label', 'ASC')
->groupBy('w.id');
return $query->execute()->fetchAll(\PDO::FETCH_ASSOC);
return $query->executeQuery()->fetchAllAssociative();
}
public function save(ScheduledDataflow $scheduledDataflow)
@@ -107,16 +91,16 @@ class ScheduledDataflowRepository
unset($datas['id']);
if (is_array($datas['options'])) {
$datas['options'] = json_encode($datas['options']);
$datas['options'] = json_encode($datas['options'], JSON_THROW_ON_ERROR);
}
if (null === $scheduledDataflow->getId()) {
$this->connection->insert(static::TABLE_NAME, $datas, static::FIELDS_TYPE);
$this->connection->insert(static::TABLE_NAME, $datas, $this->getFields());
$scheduledDataflow->setId((int) $this->connection->lastInsertId());
return;
}
$this->connection->update(static::TABLE_NAME, $datas, ['id' => $scheduledDataflow->getId()], static::FIELDS_TYPE);
$this->connection->update(static::TABLE_NAME, $datas, ['id' => $scheduledDataflow->getId()], $this->getFields());
}
public function delete(int $id): void
@@ -144,11 +128,24 @@ class ScheduledDataflowRepository
private function returnFirstOrNull(QueryBuilder $qb): ?ScheduledDataflow
{
$stmt = $qb->execute();
$stmt = $qb->executeQuery();
if (0 === $stmt->rowCount()) {
return null;
}
return ScheduledDataflow::createFromArray($this->initDateTime($this->initArray($stmt->fetch(\PDO::FETCH_ASSOC))));
return ScheduledDataflow::createFromArray($this->initDateTime($this->initArray($stmt->fetchAssociative())));
}
private function getFields(): array
{
return [
'id' => ParameterType::INTEGER,
'label' => ParameterType::STRING,
'dataflow_type' => ParameterType::STRING,
'options' => ParameterType::STRING,
'frequency' => ParameterType::STRING,
'next' => 'datetime',
'enabled' => ParameterType::BOOLEAN,
];
}
}

View File

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

@@ -45,9 +45,15 @@ services:
tags: ['console.command']
CodeRhapsodie\DataflowBundle\Command\SchemaCommand:
deprecated:
package: 'code-rhapsodie/dataflow-bundle'
version: '5.0'
tags: ['console.command']
CodeRhapsodie\DataflowBundle\Command\DatabaseSchemaCommand:
arguments:
$connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
tags: ['console.command']
tags: [ 'console.command' ]
CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository:
lazy: true
@@ -76,6 +82,12 @@ services:
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:
$repository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
$registry: '@CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface'

View File

@@ -0,0 +1,26 @@
<?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
{
public function __construct(private JobRepository $repository, private MessageBusInterface $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,30 +4,13 @@ declare(strict_types=1);
namespace CodeRhapsodie\DataflowBundle\Runner;
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\Processor\JobProcessorInterface;
use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;
class PendingDataflowRunner implements PendingDataflowRunnerInterface
{
/** @var JobRepository */
private $repository;
/** @var DataflowTypeRegistryInterface */
private $registry;
/** @var EventDispatcherInterface */
private $dispatcher;
public function __construct(JobRepository $repository, DataflowTypeRegistryInterface $registry, EventDispatcherInterface $dispatcher)
public function __construct(private JobRepository $repository, private JobProcessorInterface $processor)
{
$this->repository = $repository;
$this->registry = $registry;
$this->dispatcher = $dispatcher;
}
/**
@@ -36,49 +19,7 @@ class PendingDataflowRunner implements PendingDataflowRunnerInterface
public function runPendingDataflows(): void
{
while (null !== ($job = $this->repository->findNextPendingDataflow())) {
$this->beforeProcessing($job);
$dataflowType = $this->registry->getDataflowType($job->getDataflowType());
$result = $dataflowType->process($job->getOptions());
$this->afterProcessing($job, $result);
$this->processor->process($job);
}
}
/**
* @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->repository->save($job);
}
/**
* @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->repository->save($job);
$this->dispatcher->dispatch(Events::AFTER_PROCESSING, new ProcessingEvent($job));
}
}

View File

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

View File

@@ -13,7 +13,7 @@ class FrequencyValidator extends ConstraintValidator
/**
* {@inheritdoc}
*/
public function validate($value, Constraint $constraint)
public function validate(mixed $value, Constraint $constraint)
{
if (!$constraint instanceof Frequency) {
throw new UnexpectedTypeException($constraint, Frequency::class);
@@ -23,7 +23,12 @@ class FrequencyValidator extends ConstraintValidator
return;
}
$interval = @\DateInterval::createFromDateString($value);
try {
$interval = \DateInterval::createFromDateString($value);
} catch (\Exception){
$interval = false;
}
if (!$interval) {
$this->context->buildViolation($constraint->message)
->setParameter('{{ string }}', $value)
@@ -42,8 +47,6 @@ class FrequencyValidator extends ConstraintValidator
->setParameter('{{ string }}', $value)
->addViolation()
;
return;
}
}
}