refacto filters

src/Command/AbstractImportCommand.php

@@ -40,6 +40,7 @@ protected function configure(): void
             ->addOption('parallel', 'p', InputOption::VALUE_NONE, 'Allow parallel task processing')
             ->addOption('disable-batch', 'd', InputOption::VALUE_NONE, 'Disable batch processing')
             ->addOption('batch-size', 's', InputOption::VALUE_OPTIONAL, 'Batch Size', 100)
+            ->addOption('from-page', null, InputOption::VALUE_OPTIONAL, 'From page', 1)
             ->addOption('max-concurrency', 'c', InputOption::VALUE_OPTIONAL, 'Max process concurrency', 5)
             ->addOption('batch-after-fetch', 'a', InputOption::VALUE_OPTIONAL, 'Fetch all pages then start processing the batches', true)
             ->addOption('filter', 'f', InputOption::VALUE_OPTIONAL | InputOption::VALUE_IS_ARRAY, 'Add filter')

src/Command/BatchImportCategoriesCommand.php

@@ -0,0 +1,48 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Synolia\SyliusAkeneoPlugin\Command;
+
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Synolia\SyliusAkeneoPlugin\Client\ClientFactoryInterface;
+use Synolia\SyliusAkeneoPlugin\Payload\Category\CategoryPayload;
+use Synolia\SyliusAkeneoPlugin\Task\Category\BatchCategoriesTask;
+
+final class BatchImportCategoriesCommand extends AbstractBatchCommand
+{
+    protected static $defaultDescription = 'Import batch categories ids from Akeneo PIM.';
+
+    /** @var string */
+    protected static $defaultName = 'akeneo:batch:categories';
+
+    public function __construct(
+        private ClientFactoryInterface $clientFactory,
+        private LoggerInterface $logger,
+        private BatchCategoriesTask $task,
+    ) {
+        parent::__construct(self::$defaultName);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function execute(
+        InputInterface $input,
+        OutputInterface $output,
+    ) {
+        $ids = explode(',', $input->getArgument('ids'));
+
+        $this->logger->notice('Processing batch', ['from_id' => $ids[0], 'to_id' => $ids[\count($ids) - 1]]);
+        $this->logger->debug(self::$defaultName, ['batched_ids' => $ids]);
+
+        $payload = new CategoryPayload($this->clientFactory->createFromApiCredentials());
+        $payload->setIds($ids);
+
+        $this->task->__invoke($payload);
+
+        return 0;
+    }
+}

src/Configuration/ConfigurationContextInterface.php

@@ -43,4 +43,8 @@ public function setFilters(array $filters): self;
     public function getHandler(): string;
 
     public function setHandler(string $handler): self;
+
+    public function getFromPage(): int;
+
+    public function setFromPage(int $fromPage): self;
 }

src/Configuration/ConfigurationContextTrait.php

@@ -27,6 +27,8 @@ trait ConfigurationContextTrait
 
     private string $handler = SymfonyProcessTaskHandler::HANDLER_CODE;
 
+    private int $fromPage = 1;
+
     public function getBatchSize(): int
     {
         return $this->batchSize;
@@ -147,4 +149,16 @@ public function setHandler(string $handler): ConfigurationContextInterface
 
         return $this;
     }
+
+    public function getFromPage(): int
+    {
+        return $this->fromPage;
+    }
+
+    public function setFromPage(int $fromPage): ConfigurationContextInterface
+    {
+        $this->fromPage = $fromPage;
+
+        return $this;
+    }
 }

src/Factory/CategoryPipelineFactory.php

@@ -8,7 +8,7 @@
 use League\Pipeline\PipelineInterface;
 use Synolia\SyliusAkeneoPlugin\Pipeline\Processor;
 use Synolia\SyliusAkeneoPlugin\Task\Category\CreateUpdateEntityTask;
-use Synolia\SyliusAkeneoPlugin\Task\Category\RetrieveCategoriesTask;
+use Synolia\SyliusAkeneoPlugin\Task\Category\ProcessCategoriesTask;
 
 final class CategoryPipelineFactory extends AbstractPipelineFactory
 {
@@ -17,7 +17,7 @@ public function create(): PipelineInterface
         $pipeline = new Pipeline(new Processor($this->dispatcher));
 
         return $pipeline
-            ->pipe($this->taskProvider->get(RetrieveCategoriesTask::class))
+            ->pipe($this->taskProvider->get(ProcessCategoriesTask::class))
             ->pipe($this->taskProvider->get(CreateUpdateEntityTask::class))
         ;
     }

src/Factory/PayloadFactory.php

@@ -51,6 +51,7 @@ private function createContext(
             ->setBatchingAllowed($isBatchingAllowed)
             ->setProcessAsSoonAsPossible(filter_var($batchAfterFetch, FILTER_VALIDATE_BOOLEAN))
             ->setBatchSize((int) $input->getOption('batch-size'))
+            ->setFromPage((int) $input->getOption('from-page'))
             ->setMaxRunningProcessQueueSize((int) $input->getOption('max-concurrency'))
             ->setFilters((array) ($input->getOption('filter') ?: []))
             ->setHandler($input->getOption('handler') ?? $context->getHandler())

src/Handler/Task/SymfonyMessengerTaskHandler.php

@@ -6,7 +6,6 @@
 
 use Akeneo\Pim\ApiClient\Pagination\Page;
 use Akeneo\Pim\ApiClient\Pagination\PageInterface;
-use Akeneo\Pim\ApiClient\Pagination\ResourceCursorInterface;
 use Doctrine\ORM\EntityManagerInterface;
 use Psr\Log\LoggerInterface;
 use Symfony\Component\Messenger\MessageBusInterface;
@@ -37,7 +36,7 @@ public function batch(PipelinePayloadInterface $pipelinePayload, array $items):
 
     public function handle(
         PipelinePayloadInterface $pipelinePayload,
-        ResourceCursorInterface|PageInterface $handleType,
+        iterable|PageInterface $handleType,
     ): void {
         $count = 0;
         $items = [];
@@ -83,7 +82,7 @@ private function handleByPage(
 
     private function handleByCursor(
         PipelinePayloadInterface $payload,
-        ResourceCursorInterface $resourceCursor,
+        iterable $resourceCursor,
         int &$count = 0,
         array &$items = [],
     ): void {

src/Handler/Task/SymfonyProcessTaskHandler.php

@@ -114,7 +114,7 @@ public function continue(PipelinePayloadInterface $pipelinePayload): void
      */
     public function handle(
         PipelinePayloadInterface $pipelinePayload,
-        ResourceCursorInterface|PageInterface $handleType,
+        iterable|PageInterface $handleType,
     ): void {
         $this->processManager->setInstantProcessing($pipelinePayload->getProcessAsSoonAsPossible());
         $this->processManager->setNumberOfParallelProcesses($pipelinePayload->getMaxRunningProcessQueueSize());

src/Handler/Task/TaskHandlerInterface.php

@@ -5,7 +5,6 @@
 namespace Synolia\SyliusAkeneoPlugin\Handler\Task;
 
 use Akeneo\Pim\ApiClient\Pagination\PageInterface;
-use Akeneo\Pim\ApiClient\Pagination\ResourceCursorInterface;
 use Symfony\Component\DependencyInjection\Attribute\AutoconfigureTag;
 use Synolia\SyliusAkeneoPlugin\Payload\PipelinePayloadInterface;
 
@@ -25,7 +24,7 @@ public function batch(
 
     public function handle(
         PipelinePayloadInterface $pipelinePayload,
-        ResourceCursorInterface|PageInterface $handleType,
+        iterable|PageInterface $handleType,
     ): void;
 
     public function continue(PipelinePayloadInterface $pipelinePayload): void;

src/Message/Batch/CategoryBatchMessage.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Synolia\SyliusAkeneoPlugin\Message\Batch;
+
+class CategoryBatchMessage implements BatchMessageInterface
+{
+    public function __construct(public array $items)
+    {
+    }
+}

src/MessageHandler/Batch/CategoryBatchMessageHandler.php

@@ -0,0 +1,35 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Synolia\SyliusAkeneoPlugin\MessageHandler\Batch;
+
+use Symfony\Component\Messenger\Attribute\AsMessageHandler;
+use Symfony\Contracts\EventDispatcher\EventDispatcherInterface;
+use Synolia\SyliusAkeneoPlugin\Message\Batch\CategoryBatchMessage;
+use Synolia\SyliusAkeneoPlugin\Processor\Resource\Category\CategoryResourceProcessor;
+use Synolia\SyliusAkeneoPlugin\Processor\Resource\Exception\MaxResourceProcessorRetryException;
+
+#[AsMessageHandler]
+class CategoryBatchMessageHandler
+{
+    public function __construct(
+        private EventDispatcherInterface $dispatcher,
+        private CategoryResourceProcessor $resourceProcessor,
+    ) {
+    }
+
+    public function __invoke(CategoryBatchMessage $attributeBatchMessage): void
+    {
+        foreach ($attributeBatchMessage->items as $resource) {
+            try {
+                $this->resourceProcessor->process($resource);
+            } catch (MaxResourceProcessorRetryException) {
+                // Skip the failing line
+                $this->dispatcher->dispatch(new CategoryBatchMessage([$resource]));
+
+                continue;
+            }
+        }
+    }
+}

src/Payload/AbstractPayload.php

@@ -39,6 +39,7 @@ public function __construct(
             $this->isContinue = $commandContext->isContinue();
             $this->processAsSoonAsPossible = $commandContext->getProcessAsSoonAsPossible();
             $this->handler = $commandContext->getHandler();
+            $this->fromPage = $commandContext->getFromPage();
         }
     }
 

src/Payload/Category/CategoryPayload.php

@@ -4,12 +4,29 @@
 
 namespace Synolia\SyliusAkeneoPlugin\Payload\Category;
 
+use Akeneo\Pim\ApiClient\AkeneoPimClientInterface;
+use Synolia\SyliusAkeneoPlugin\Command\Context\CommandContextInterface;
 use Synolia\SyliusAkeneoPlugin\Exceptions\NoCategoryResourcesException;
 use Synolia\SyliusAkeneoPlugin\Message\Batch\BatchMessageInterface;
+use Synolia\SyliusAkeneoPlugin\Message\Batch\CategoryBatchMessage;
 use Synolia\SyliusAkeneoPlugin\Payload\AbstractPayload;
 
 final class CategoryPayload extends AbstractPayload
 {
+    public const TEMP_AKENEO_TABLE_NAME = 'tmp_akeneo_categories';
+
+    public const BATCH_COMMAND_NAME = 'akeneo:batch:categories';
+
+    public function __construct(
+        AkeneoPimClientInterface $akeneoPimClient,
+        ?CommandContextInterface $commandContext = null,
+    ) {
+        parent::__construct($akeneoPimClient, $commandContext);
+
+        $this->setTmpTableName(self::TEMP_AKENEO_TABLE_NAME);
+        $this->setCommandName(self::BATCH_COMMAND_NAME);
+    }
+
     private array $resources;
 
     public function getResources(): array
@@ -28,6 +45,6 @@ public function setResources(array $resources): void
 
     public function createBatchMessage(array $items): BatchMessageInterface
     {
-        throw new \InvalidArgumentException();
+        return new CategoryBatchMessage($items);
     }
 }