Extending

This guide presents a solution for efficiently managing resource-intensive, sequential tasks within the CoreShop Batch Messenger Bundle, specifically tailored for Pimcore. Our focus is on creating a streamlined process to generate image previews from PDF files and store them as assets. The implementation encompasses several crucial steps:

1. Defining a New Task Type: We introduce PdfPreviewTask to encapsulate the necessary data for task processing, including productId and pageNumber.

2. Developing a Task Item Processor: A specialized processor is developed to handle individual Task Items. Given that PDFs are stored externally, we utilize the Flysystem FilesystemOperator.

3. Processor Registration: The processor is registered as a service and tagged as an ItemProcessor within the Batch Messenger system, ensuring correct recognition and utilization.

4. Automating Task Creation and Dispatching: The createTask method is implemented to facilitate the automatic generation of new tasks, each comprising a specified number of Task Items. These tasks are then dispatched to the Messenger, which processes them in efficient batches.

The motivation behind this development is to optimize the handling of large-scale, sequential operations in Pimcore. By employing task batching and a sophisticated processing system, we aim to reduce resource consumption and enhance operational efficiency, especially for demanding tasks like generating PDF previews. This approach is designed to improve scalability and performance in Pimcore's dynamic environment.

Create a new Task Type

Define a TaskType to hold necessary data for processing TaskItems. For this example, we need the productId and pageNumber.

<?php

namespace App\CoreShop\BatchMessenger;

use CoreShop\Bundle\BatchMessengerBundle\Task\TaskTypeInterface;

class PdfPreviewTask implements TaskTypeInterface
{
    public function __construct(
        public readonly int $productId,
        public readonly int $page = 1,
    )
    {
    }
}

Create the Task Item Processor

Create a processor to handle TaskItem. This example assumes PDFs are stored externally, hence the use of Flysystem FilesystemOperator.

<?php

namespace App\CoreShop\BatchMessenger\Processor;

use App\CoreShop\BatchMessenger\PdfPreviewTask;
use CoreShop\Bundle\BatchMessengerBundle\Model\TaskItemInterface;
use CoreShop\Bundle\BatchMessengerBundle\Processor\TaskItemProcessorInterface;
use League\Flysystem\FilesystemOperator;
use Pimcore\File;
use Pimcore\Model\Asset\Folder;
use Pimcore\Model\Asset\Image;
use Pimcore\Model\Asset\Service;
use Pimcore\Model\DataObject\Product;

class PdfPreviewProcessor implements TaskItemProcessorInterface
{
    public function __construct(
        private readonly FilesystemOperator $pdfStorage
    ) {
    }

    public function supports(TaskItemInterface $taskItem): bool
    {
        return $taskItem->getTaskData() instanceof PdfPreviewTask;
    }

    public function processItem(TaskItemInterface $taskItem, array $context): void
    {
        /**
         * @var PdfPreviewTask $taskData
         */
        $taskData = $taskItem->getTaskData();

        $product = Product::getById($taskData->productId);

        if (!$product) {
            return;
        }

        $page = $taskData->page;
        $previewsFolder = $product->getPdfPreviewDocuments();

        if (!$previewsFolder) {
            return;
        }

        $outputImageName = $product->getId().'-'.$page.'.png';
        $fullImagePath = sprintf('%s/%s', $previewsFolder->getFullPath(), Service::getValidKey($outputImageName, 'asset'));
        $imageAsset = Image::getByPath($fullImagePath);

        if ($imageAsset) {
            return;
        }

        $pdfPath = $product->getPdfPath();

        if (!$pdfPath) {
            return;
        }

        if (!$this->pdfStorage->fileExists($pdfPath)) {
            return;
        }

        $pdfStream = $this->pdfStorage->readStream($pdfPath);

        $tempFile = File::getLocalTempFilePath();

        file_put_contents($tempFile, stream_get_contents($pdfStream));

        $imagick = new \Imagick();
        $imagick->readImage($tempFile);
        $imagick->setResolution(300, 300);
        $totalPdfPages = $imagick->getNumberImages();

        if ($page > $totalPdfPages) {
            return;
        }

        $tempFileImage = File::getLocalTempFilePath();
        $imagick->readImage($tempFile.'['.$page.']');
        $tempImagePath = $tempFileImage.'.png';
        $imagick->writeImage($tempImagePath);

        $imageAsset = new Image();
        $imageAsset->setParent($previewsFolder);
        $imageAsset->setFilename(Service::getValidKey($outputImageName, 'asset'));
        $imageAsset->setData(file_get_contents($tempImagePath));
        $imageAsset->save();
    }
}

Register the Processor

Register the Processor as a service and tag it as an ItemProcessor.

services:
  _defaults:
    autowire: true
    autoconfigure: true
    public: false

  App\CoreShop\BatchMessenger\Processor\PdfPreviewProcessor:
    tags:
      - { name: coreshop_batch_messenger.item_processor }

Task Creator

The createTask method generates a new task with the specified number of TaskItems and sends it to the Messenger, which processes them in batches.

<?php

namespace App\CoreShop\BatchMessenger;

use CoreShop\Bundle\BatchMessengerBundle\Creator\JobCreatorInterface;
use CoreShop\Bundle\BatchMessengerBundle\Factory\TaskFactoryInterface;
use CoreShop\Bundle\BatchMessengerBundle\Factory\TaskItemFactoryInterface;
use CoreShop\Bundle\BatchMessengerBundle\Messenger\TaskMessage;
use CoreShop\Bundle\BatchMessengerBundle\Model\TaskInterface;
use CoreShop\Bundle\BatchMessengerBundle\Task\TaskTypeInterface;
use Doctrine\ORM\EntityManagerInterface;
use Pimcore\Model\DataObject\Product;
use Symfony\Component\Messenger\MessageBusInterface;

class PreviewJobCreator implements JobCreatorInterface
{
    public function __construct(
        private readonly TaskFactoryInterface $factory,
        private readonly TaskItemFactoryInterface $itemFactory,
        private readonly EntityManagerInterface $entityManager,
        private readonly MessageBusInterface $bus,
    ) {
    }

    public function createTask(Product $product): TaskInterface
    {
        $tasks = [];
        
        for ($i = 0; $i < $countPages; $i++) {
            $tasks[] = new PdfPreviewTask($product->getId(), $i);
        }

        $taskItems = array_map(function (TaskTypeInterface $task) {
            return $this->itemFactory->create($task);
        }, $tasks);

        $task = $this->factory->createWithItems(
            $taskItems,
        );
        //Set how many are processed within one Messenger Job
        $task->setBatchSize(3);

        $this->entityManager->persist($task);
        $this->entityManager->flush();

        $this->bus->dispatch(
            new TaskMessage((int)$task->getId()),
        );

        return $task;
    }
}

Now, invoking the createTask method initiates a new Task with the specified number of Task Items. These items are dispatched to the Messenger for processing, which handles them in batches of three for optimized execution.