Concrete CMS Batch Processing Command

BatchProcessingCommand.php

<?php
/**
 * @author: Biplob Hossain <biplob.ice@gmail.com>
 * @license MIT
 */

namespace BatchProcessing\Console\Command;

use Concrete\Core\Console\Command;
use Concrete\Core\Support\Facade\Facade;
use Doctrine\ORM\EntityManagerInterface;
use Doctrine\ORM\Internal\Hydration\IterableResult;
use Doctrine\ORM\QueryBuilder;
use Symfony\Component\Console\Helper\ProgressBar;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;

abstract class BatchProcessingCommand extends Command
{
    protected $batchSize = 10000;
    protected $maxResult = 10000;

    /** @var EntityManagerInterface */
    protected $em;
    /** @var ProgressBar */
    protected $progress;

    abstract protected function getQueryBuilder(): QueryBuilder;
    abstract protected function getNumOfTotalRecords(): int;
    abstract protected function processRecord(object $record);

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $returnCode = 0;
        $this->input = $input;
        $this->output = $output;

        // Return if no records are found to delete
        $totalRecords = $this->getNumOfTotalRecords();
        if (!$totalRecords) {
            $this->output->writeln(t('No records found to process.'));

            return $returnCode;
        }

        // Set the progress bar
        $this->startProgressBar($output, $totalRecords);

        $totalProcessed = 0;
        $processing = true;

        while ($processing) {
            $iterableResult = $this->getIterableResult($totalProcessed, $this->maxResult);

            while (($row = $iterableResult->next()) !== false) {
                $this->processRecord($row[0]);
                $this->progress->advance();

                if (($totalProcessed % $this->batchSize ) === 0) {
                    $this->onBatchComplete();
                }
                $totalProcessed++;
            }
            if ($totalProcessed === $totalRecords) {
                $processing = false;
            }
        }

        $this->progress->finish();

        return $returnCode;
    }

    protected function getEntityManager()
    {
        if (!$this->em) {
            $this->em = Facade::getFacadeApplication()->make(EntityManagerInterface::class);
        }

        return $this->em;
    }

    protected function startProgressBar(OutputInterface $output, int $totalRecords): void
    {
        $this->progress = new ProgressBar($output, $totalRecords);
        $this->progress->setFormat(
            implode(
                "\n",
                [
                    '<fg=green;options=bold>' . t('Processing Record') . ' </> %current%/%max%',
                    '    <info>Progress : %percent:3s%%</info>',
                    '    <fg=cyan;options=reverse>[%bar%]</>',
                    '    <comment>%elapsed:6s% / %estimated:-6s% %memory:6s%</comment>',
                ]
            )
        );

        $this->progress->start();
    }

    protected function getIterableResult(int $firstResult = 0, int $maxResult = 0): IterableResult
    {
        $qb = $this->getQueryBuilder();
        if ($firstResult > 0) {
            $qb->setFirstResult($firstResult);
        }

        if ($maxResult > 0) {
            $qb->setMaxResults($maxResult);
        }
        return $qb->getQuery()->iterate();
    }
    
    protected function onBatchComplete(): void
    {
        $this->getEntityManager()->clear();
    }    
}