Async processor must be independent classes

src/Bundle/Command/AbstractImportCommand.php

@@ -15,7 +15,6 @@
 
 use IQ2i\DataImporter\Archiver\ArchiverInterface;
 use IQ2i\DataImporter\Bundle\Exception\ItemHandlingException;
-use IQ2i\DataImporter\Bundle\Processor\AsyncCliProcessor;
 use IQ2i\DataImporter\Bundle\Processor\CliProcessor;
 use IQ2i\DataImporter\DataImporter;
 use IQ2i\DataImporter\Exchange\Message;
@@ -43,7 +42,6 @@ protected function configure(): void
     {
         $this
             ->addArgument('filename', InputArgument::OPTIONAL, 'File to import')
-            ->addOption('async', null, InputOption::VALUE_NONE, 'Parallelize row process using symfony/messenger')
             ->addOption('step', null, InputOption::VALUE_NONE, 'Step through each record one-by-one')
             ->addOption('pause-on-error', null, InputOption::VALUE_NONE, 'Pause if an exception is thrown')
             ->addOption('batch-size', null, InputOption::VALUE_REQUIRED, 'Batch size', 100)
@@ -95,9 +93,7 @@ protected function handleEnd(): callable
 
     protected function getProcessor(InputInterface $input, OutputInterface $output): ProcessorInterface
     {
-        return $input->getOption('async')
-            ? new AsyncCliProcessor($input, $output, $this->handleBegin(), $this->handleItem(), $this->handleBatch(), $this->handleEnd(), $this->getSerializer())
-            : new CliProcessor($input, $output, $this->handleBegin(), $this->handleItem(), $this->handleBatch(), $this->handleEnd(), $this->getSerializer());
+        return new CliProcessor($input, $output, $this->handleBegin(), $this->handleItem(), $this->handleBatch(), $this->handleEnd(), $this->getSerializer());
     }
 
     protected function getArchiver(): ?ArchiverInterface

src/Bundle/Messenger/ProcessItemMessage.php → src/Bundle/Messenger/AsyncMessage.php

@@ -15,19 +15,13 @@
 
 use IQ2i\DataImporter\Exchange\Message;
 
-class ProcessItemMessage
+class AsyncMessage
 {
     public function __construct(
-        private readonly \Closure $handleItem,
         private readonly Message $data,
     ) {
     }
 
-    public function getHandleItem(): \Closure
-    {
-        return $this->handleItem;
-    }
-
     public function getData(): Message
     {
         return $this->data;

src/Bundle/Processor/AbstractAsyncProcessor.php

@@ -0,0 +1,46 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the DataImporter package.
+ *
+ * (c) Loïc Sapone <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace IQ2i\DataImporter\Bundle\Processor;
+
+use IQ2i\DataImporter\Bundle\Messenger\AsyncMessage;
+use IQ2i\DataImporter\Exchange\Message;
+use IQ2i\DataImporter\Processor\ProcessorInterface;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Symfony\Contracts\Service\Attribute\Required;
+
+abstract class AbstractAsyncProcessor implements ProcessorInterface
+{
+    private ?MessageBusInterface $bus = null;
+
+    public function begin(Message $message)
+    {
+    }
+
+    public function item(Message $message)
+    {
+        $this->bus->dispatch(new AsyncMessage($message));
+    }
+
+    abstract public function processItem(AsyncMessage $message);
+
+    public function end(Message $message)
+    {
+    }
+
+    #[Required]
+    public function setBus(MessageBusInterface $bus): void
+    {
+        $this->bus = $bus;
+    }
+}

src/Bundle/Resources/config/services.php

@@ -13,7 +13,6 @@
 
 namespace Symfony\Component\DependencyInjection\Loader\Configurator;
 
-use IQ2i\DataImporter\Bundle\Messenger\MessageHandler;
 use IQ2i\DataImporter\Command\GenerateDtoCommand;
 
 return static function (ContainerConfigurator $container) {
@@ -22,8 +21,5 @@
             ->arg('$defaultPath', '%kernel.project_dir%')
             ->arg('$defaultNamespace', 'App\\Dto')
             ->tag('console.command', ['command' => 'iq2i:data-importer:generate-dto'])
-
-        ->set('iq2i_data_importer.messenger.message_handler', MessageHandler::class)
-            ->tag('messenger.message_handler')
     ;
 };

src/DataImporter.php

@@ -14,14 +14,11 @@
 namespace IQ2i\DataImporter;
 
 use IQ2i\DataImporter\Archiver\ArchiverInterface;
-use IQ2i\DataImporter\Bundle\Messenger\ProcessItemMessage;
 use IQ2i\DataImporter\Exchange\MessageFactory;
-use IQ2i\DataImporter\Processor\AsyncProcessorInterface;
 use IQ2i\DataImporter\Processor\BatchProcessorInterface;
 use IQ2i\DataImporter\Processor\ProcessorInterface;
 use IQ2i\DataImporter\Reader\ReaderInterface;
 use Symfony\Component\Filesystem\Exception\IOException;
-use Symfony\Component\Messenger\MessageBusInterface;
 use Symfony\Component\Serializer\NameConverter\CamelCaseToSnakeCaseNameConverter;
 use Symfony\Component\Serializer\Normalizer\ObjectNormalizer;
 use Symfony\Component\Serializer\Serializer;
@@ -36,7 +33,6 @@ public function __construct(
         private readonly ProcessorInterface $processor,
         private readonly ?ArchiverInterface $archiver = null,
         SerializerInterface $serializer = null,
-        private readonly ?MessageBusInterface $bus = null,
     ) {
         $this->serializer = $serializer ?? new Serializer([new ObjectNormalizer(null, new CamelCaseToSnakeCaseNameConverter())]);
     }
@@ -51,12 +47,6 @@ public function execute(): void
                 $this->reader->isDenormalizable() ? $this->serializeData($data) : $data
             );
 
-            if ($this->processor instanceof AsyncProcessorInterface && null !== $this->bus) {
-                $this->bus->dispatch(new ProcessItemMessage(fn () => $this->processor->item($message), $message));
-
-                continue;
-            }
-
             $this->processor->item($message);
 
             if ($this->processor instanceof BatchProcessorInterface && (

tests/DataImporterTest.php

@@ -14,7 +14,7 @@
 namespace IQ2i\DataImporter\Tests;
 
 use IQ2i\DataImporter\Archiver\DateTimeArchiver;
-use IQ2i\DataImporter\Bundle\Messenger\ProcessItemMessage;
+use IQ2i\DataImporter\Bundle\Messenger\AsyncMessage;
 use IQ2i\DataImporter\DataImporter;
 use IQ2i\DataImporter\Processor\CallbackProcessor;
 use IQ2i\DataImporter\Reader\CsvReader;
@@ -168,9 +168,9 @@ public function testWithUnreadableArchivePath()
     public function testAsyncProcessor()
     {
         $bus = new class() implements MessageBusInterface {
-            public $messages = [];
+            public array $messages = [];
 
-            public $stamps = [];
+            public array $stamps = [];
 
             public function dispatch($message, array $stamps = []): Envelope
             {
@@ -181,26 +181,25 @@ public function dispatch($message, array $stamps = []): Envelope
             }
         };
 
+        $processor = new AsyncProcessor();
+        $processor->setBus($bus);
+
         $dataImporter = new DataImporter(
             new CsvReader(
                 $this->fs->getChild('books.csv')->url(),
                 null,
                 [CsvReader::CONTEXT_DELIMITER => ';']
             ),
-            new AsyncProcessor(),
-            null,
-            null,
-            $bus
+            $processor,
         );
 
         $dataImporter->execute();
 
         $this->assertCount(2, $bus->messages);
 
-        /** @var ProcessItemMessage $message */
+        /** @var AsyncMessage $message */
         $message = \array_pop($bus->messages);
 
-        $this->assertEquals(static function () {}, $message->getHandleItem());
         $this->assertEquals([
             'author' => 'Ralls, Kim',
             'title' => 'Midnight Rain',

tests/fixtures/Processor/AsyncProcessor.php

@@ -13,21 +13,12 @@
 
 namespace IQ2i\DataImporter\Tests\fixtures\Processor;
 
-use IQ2i\DataImporter\Exchange\Message;
-use IQ2i\DataImporter\Processor\AsyncProcessorInterface;
-use IQ2i\DataImporter\Processor\ProcessorInterface;
+use IQ2i\DataImporter\Bundle\Messenger\AsyncMessage;
+use IQ2i\DataImporter\Bundle\Processor\AbstractAsyncProcessor;
 
-class AsyncProcessor implements ProcessorInterface, AsyncProcessorInterface
+class AsyncProcessor extends AbstractAsyncProcessor
 {
-    public function begin(Message $message)
-    {
-    }
-
-    public function item(Message $message)
-    {
-    }
-
-    public function end(Message $message)
+    public function processItem(AsyncMessage $message)
     {
     }
 }