Add auto update count processed item while running job

.github/workflows/ci.yml

@@ -10,3 +10,5 @@ jobs:
       - uses: actions/checkout@v4
       - uses: php-actions/composer@v6 # or alternative dependency management
       - uses: php-actions/phpunit@v4
+      - name: Run PHP CS Fixer
+        run: php vendor/bin/php-cs-fixer fix --dry-run --diff

.gitignore

@@ -6,3 +6,4 @@ composer.lock
 .idea
 .phpunit.cache
 .php-version
+.php-cs-fixer.cache

.php-cs-fixer.dist.php

@@ -0,0 +1,49 @@
+<?php
+
+$finder = PhpCsFixer\Finder::create()
+    ->in('src')
+    //->in('tests')
+    ->files()->name('*.php');
+
+$config = new PhpCsFixer\Config();
+$config->setRules([
+    '@Symfony' => true,
+    '@Symfony:risky' => true,
+    '@PSR12' => true,
+    'array_syntax' => [
+        'syntax' => 'short',
+    ],
+    'combine_consecutive_unsets' => true,
+    'native_function_invocation' => [
+        'include' => [
+            '@compiler_optimized',
+        ],
+    ],
+    'no_extra_blank_lines' => [
+        'tokens' => [
+            'break',
+            'continue',
+            'extra',
+            'return',
+            'throw',
+            'use',
+            'parenthesis_brace_block',
+            'square_brace_block',
+            'curly_brace_block',
+        ],
+    ],
+    'ordered_class_elements' => true,
+    'ordered_imports' => true,
+    'yoda_style' => [
+        'equal' => false,
+        'identical' => false,
+        'less_and_greater' => false,
+        'always_move_variable' => false,
+    ],
+])
+    ->setRiskyAllowed(true)
+    ->setFinder(
+        $finder
+    );
+
+return $config;

CHANGELOG.md

@@ -1,3 +1,6 @@
+# Version 5.3.0
+* Added auto update count processed item while running job
+
 # Version 5.2.0
 * Added custom index for job status
 

Tests/DataflowType/Writer/CollectionWriterTest.php

@@ -44,9 +44,7 @@ public function testAll()
         $embeddedWriter
             ->expects($matcher)
             ->method('write')
-            ->with($this->callback(function ($arg) use ($matcher, $values) {
-                return $arg === $values[$matcher->numberOfInvocations() - 1];
-            }))
+            ->with($this->callback(fn($arg) => $arg === $values[$matcher->numberOfInvocations() - 1]))
         ;
 
         $writer = new CollectionWriter($embeddedWriter);

Tests/DataflowType/Writer/PortWriterAdapterTest.php

@@ -11,7 +11,7 @@ public function testAll()
     {
         $value = 'not an array';
 
-        $writer = $this->getMockBuilder('\Port\Writer')
+        $writer = $this->getMockBuilder(\Port\Writer::class)
             ->onlyMethods(['prepare', 'finish', 'writeItem'])
             ->getMock()
         ;

Tests/Manager/ScheduledDataflowManagerTest.php

@@ -49,15 +49,10 @@ public function testCreateJobsFromScheduledDataflows()
         $this->jobRepository
             ->expects($matcher)
             ->method('findPendingForScheduledDataflow')
-            ->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;
-                }
+            ->with($this->callback(fn($arg) => match ($matcher->numberOfInvocations()) {
+                1 => $arg === $scheduled1,
+                2 => $arg === $scheduled2,
+                default => false,
             }))
             ->willReturnOnConsecutiveCalls(new Job(), null)
         ;

Tests/Processor/JobProcessorTest.php

@@ -44,18 +44,11 @@ public function testProcess()
             ->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;
-                    }
+                $this->callback(fn($arg) => $arg instanceof ProcessingEvent && $arg->getJob() === $job),
+                $this->callback(fn($arg) => match ($matcher->numberOfInvocations()) {
+                    1 => $arg === Events::BEFORE_PROCESSING,
+                    2 => $arg === Events::AFTER_PROCESSING,
+                    default => false,
                 })
             );
 

Tests/Runner/MessengerDataflowRunnerTest.php

@@ -39,31 +39,21 @@ public function testRunPendingDataflows()
         $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;
-                }
+            ->with($this->callback(fn($arg) => match ($matcher->numberOfInvocations()) {
+                1 => $arg === $job1,
+                2 => $arg === $job2,
+                default => 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;
-                }
+            ->with($this->callback(fn($arg) => match ($matcher->numberOfInvocations()) {
+                1 => $arg instanceof JobMessage && $arg->getJobId() === $id1,
+                2 => $arg instanceof JobMessage && $arg->getJobId() === $id2,
+                default => false,
             }))
             ->willReturnOnConsecutiveCalls(
                 new Envelope(new JobMessage($id1)),

Tests/Runner/PendingDataflowRunnerTest.php

@@ -38,15 +38,10 @@ public function testRunPendingDataflows()
         $this->processor
             ->expects($matcher)
             ->method('process')
-            ->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;
-                }
+            ->with($this->callback(fn($arg) => match ($matcher->numberOfInvocations()) {
+                1 => $arg === $job1,
+                2 => $arg === $job2,
+                default => false,
             }))
         ;
 

composer.json

@@ -60,9 +60,10 @@
     },
     "require-dev": {
         "amphp/amp": "^2.5",
+        "friendsofphp/php-cs-fixer": "^3.75",
         "phpunit/phpunit": "^11",
         "portphp/portphp": "^1.9",
-        "rector/rector": "^1.0",
+        "rector/rector": "^2.0",
         "symfony/messenger": "^7.0"
     },
     "suggest": {

rector.php

@@ -17,7 +17,7 @@
     $rectorConfig->rule(InlineConstructorDefaultToPropertyRector::class);
 
     $rectorConfig->sets([
-        SymfonySetList::SYMFONY_60,
+        SymfonySetList::SYMFONY_70,
         SymfonySetList::SYMFONY_CODE_QUALITY,
         SymfonySetList::SYMFONY_CONSTRUCTOR_INJECTION,
         LevelSetList::UP_TO_PHP_80,

src/Command/AddScheduledDataflowCommand.php

@@ -27,29 +27,27 @@ public function __construct(private DataflowTypeRegistryInterface $registry, pri
         parent::__construct();
     }
 
-    /**
-     * {@inheritdoc}
-     */
     protected function configure(): void
     {
         $this
             ->setHelp('The <info>%command.name%</info> allows you to create a new scheduled dataflow.')
             ->addOption('label', null, InputOption::VALUE_REQUIRED, 'Label of the scheduled dataflow')
             ->addOption('type', null, InputOption::VALUE_REQUIRED, 'Type of the scheduled dataflow (FQCN)')
-            ->addOption('options', null, InputOption::VALUE_OPTIONAL,
-                'Options of the scheduled dataflow (ex: {"option1": "value1", "option2": "value2"})')
+            ->addOption(
+                'options',
+                null,
+                InputOption::VALUE_OPTIONAL,
+                'Options of the scheduled dataflow (ex: {"option1": "value1", "option2": "value2"})'
+            )
             ->addOption('frequency', null, InputOption::VALUE_REQUIRED, 'Frequency of the scheduled dataflow')
             ->addOption('first_run', null, InputOption::VALUE_REQUIRED, 'Date for the first run of the scheduled dataflow (Y-m-d H:i:s)')
             ->addOption('enabled', null, InputOption::VALUE_REQUIRED, 'State of the scheduled dataflow')
             ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
     }
 
-    /**
-     * {@inheritdoc}
-     */
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
-        if (null !== $input->getOption('connection')) {
+        if ($input->getOption('connection') !== null) {
             $this->connectionFactory->setConnectionName($input->getOption('connection'));
         }
         $choices = [];
@@ -71,13 +69,17 @@ protected function execute(InputInterface $input, OutputInterface $output): int
         }
         $options = $input->getOption('options');
         if (!$options) {
-            $options = $io->ask('What are the launch options for the scheduled dataflow? (ex: {"option1": "value1", "option2": "value2"})',
-                json_encode([]));
+            $options = $io->ask(
+                'What are the launch options for the scheduled dataflow? (ex: {"option1": "value1", "option2": "value2"})',
+                json_encode([])
+            );
         }
         $frequency = $input->getOption('frequency');
         if (!$frequency) {
-            $frequency = $io->choice('What is the frequency for the scheduled dataflow?',
-                ScheduledDataflow::AVAILABLE_FREQUENCIES);
+            $frequency = $io->choice(
+                'What is the frequency for the scheduled dataflow?',
+                ScheduledDataflow::AVAILABLE_FREQUENCIES
+            );
         }
         $firstRun = $input->getOption('first_run');
         if (!$firstRun) {
@@ -92,22 +94,25 @@ protected function execute(InputInterface $input, OutputInterface $output): int
             'id' => null,
             'label' => $label,
             'dataflow_type' => $type,
-            'options' => json_decode($options, true, 512, JSON_THROW_ON_ERROR),
+            'options' => json_decode($options, true, 512, \JSON_THROW_ON_ERROR),
             'frequency' => $frequency,
             'next' => new \DateTime($firstRun),
             'enabled' => $enabled,
         ]);
 
         $errors = $this->validator->validate($newScheduledDataflow);
-        if (count($errors) > 0) {
+        if (\count($errors) > 0) {
             $io->error((string) $errors);
 
             return 2;
         }
 
         $this->scheduledDataflowRepository->save($newScheduledDataflow);
-        $io->success(sprintf('New scheduled dataflow "%s" (id:%d) was created successfully.',
-            $newScheduledDataflow->getLabel(), $newScheduledDataflow->getId()));
+        $io->success(\sprintf(
+            'New scheduled dataflow "%s" (id:%d) was created successfully.',
+            $newScheduledDataflow->getLabel(),
+            $newScheduledDataflow->getId()
+        ));
 
         return 0;
     }

src/Command/ChangeScheduleStatusCommand.php

@@ -26,9 +26,6 @@ public function __construct(private ScheduledDataflowRepository $scheduledDatafl
         parent::__construct();
     }
 
-    /**
-     * {@inheritdoc}
-     */
     protected function configure(): void
     {
         $this
@@ -39,20 +36,17 @@ protected function configure(): void
             ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
     }
 
-    /**
-     * {@inheritdoc}
-     */
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
-        if (null !== $input->getOption('connection')) {
+        if ($input->getOption('connection') !== null) {
             $this->connectionFactory->setConnectionName($input->getOption('connection'));
         }
         $io = new SymfonyStyle($input, $output);
         /** @var ScheduledDataflow|null $schedule */
         $schedule = $this->scheduledDataflowRepository->find((int) $input->getArgument('schedule-id'));
 
         if (!$schedule) {
-            $io->error(sprintf('Cannot find scheduled dataflow with id "%d".', $input->getArgument('schedule-id')));
+            $io->error(\sprintf('Cannot find scheduled dataflow with id "%d".', $input->getArgument('schedule-id')));
 
             return 1;
         }
@@ -71,9 +65,9 @@ protected function execute(InputInterface $input, OutputInterface $output): int
         try {
             $schedule->setEnabled($input->getOption('enable'));
             $this->scheduledDataflowRepository->save($schedule);
-            $io->success(sprintf('Schedule with id "%s" has been successfully updated.', $schedule->getId()));
+            $io->success(\sprintf('Schedule with id "%s" has been successfully updated.', $schedule->getId()));
         } catch (\Exception $e) {
-            $io->error(sprintf('An error occured when changing schedule status : "%s".', $e->getMessage()));
+            $io->error(\sprintf('An error occured when changing schedule status : "%s".', $e->getMessage()));
 
             return 4;
         }