eZ\Bundle\EzPublishCoreBundle\Command\ReindexCommand

Complexity

Total Complexity

38

Size/Duplication

Total Lines	281
Duplicated Lines	2.49 %

Coupling/Cohesion

Components	2
Dependencies	11

Importance

Changes

0

10 Methods

Rating	Name	Duplication	Size	Complexity
A	initialize()	0	11	2
A	configure()	0	22	1
B	execute()	0	31	4
C	indexIteratively()	0	44	7
C	runParallelProcess()	0	36	8
A	getStatementContentSince()	0	12	1
A	getStatementContentAll()	0	10	1
A	fetchIteration()	7	15	4
A	getPhpProcess()	0	13	4
B	getNumberOfCPUCores()	0	26	6

<?php

/**
 * This file is part of the eZ Publish Kernel package.
 *
 * @copyright Copyright (C) eZ Systems AS. All rights reserved.
 * @license For full copyright and license information view LICENSE file distributed with this source code.
 */
namespace eZ\Bundle\EzPublishCoreBundle\Command;

use eZ\Publish\Core\Search\Common\IterativelyIndexer;
use eZ\Publish\SPI\Persistence\Content\ContentInfo;
use eZ\Publish\Core\Search\Common\Indexer;
use Doctrine\DBAL\Driver\Statement;
use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand;
use Symfony\Component\Console\Helper\ProgressBar;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Process\Process;
use Symfony\Component\Process\PhpExecutableFinder;
use RuntimeException;
use DateTime;
use PDO;

class ReindexCommand extends ContainerAwareCommand
{
    /**
     * @var \eZ\Publish\Core\Search\Common\Indexer|\eZ\Publish\Core\Search\Common\IterativelyIndexer
     */
    private $searchIndexer;

    /**
     * @var \Doctrine\DBAL\Connection
     */
    private $connection;

    /**
     * Initialize objects required by {@see execute()}.
     *
     * @param InputInterface $input
     * @param OutputInterface $output
     */
    public function initialize(InputInterface $input, OutputInterface $output)
    {
        parent::initialize($input, $output);
        $this->searchIndexer = $this->getContainer()->get('ezpublish.spi.search.indexer');
        $this->connection = $this->getContainer()->get('ezpublish.api.storage_engine.legacy.connection');
        if (!$this->searchIndexer instanceof Indexer) {
            throw new RuntimeException(
                sprintf('Expected to find Search Engine Indexer but found "%s" instead', get_parent_class($this->searchIndexer))
            );
        }
    }

    /**
     * {@inheritdoc}
     */
    protected function configure()
    {
        $cpuCores = $this->getNumberOfCPUCores();
        $this
            ->setName('ezplatform:reindex')
            ->setDescription('Recreate or Refresh search engine index')
            ->addOption('iteration-count', 'c', InputOption::VALUE_OPTIONAL, 'Number of objects to be indexed in a single iteration, for avoiding using to much memory', 50)
            ->addOption('no-commit', null, InputOption::VALUE_NONE, 'Do not commit after each iteration')
            ->addOption('no-purge', null, InputOption::VALUE_NONE, 'Do not purge before indexing. BC NOTE: Should this be default as of 2.0?')
            ->addOption('since', null, InputOption::VALUE_OPTIONAL, 'Index changes since a given time, any format understood by DateTime. Implies "no-purge", can not be combined with "content-ids".')
            ->addOption('content-ids', null, InputOption::VALUE_IS_ARRAY | InputOption::VALUE_OPTIONAL, 'Comma separated list of content id\'s to refresh (deleted or updated/added). Implies "no-purge", can not be combined with "since".')
            ->addOption('processes', null, InputOption::VALUE_OPTIONAL, "Number of sub processes to spawn in parallel, suggested number is number of CPU cores ($cpuCores) -1")
            ->setHelp(
                <<<EOT
The command <info>%command.name%</info> indexes current configured database in configured search engine index.


TODO: ADD EXAMPLES OF ADVANCE USAGE!

EOT
            );
    }

    /**
     * {@inheritdoc}
     */
    protected function execute(InputInterface $input, OutputInterface $output)
    {
        $commit = !$input->getOption('no-commit');
        $iterationCount = $input->getOption('iteration-count');
        if (!is_numeric($iterationCount) || (int) $iterationCount < 1) {
            throw new RuntimeException("'--iteration-count' option should be > 0, got '{$iterationCount}'");
        }


        if (!$this->searchIndexer instanceof IterativelyIndexer) {
            $output->writeln( <<<EOT
DEPRECATED:
Running indexing against an Indexer that has not been updated to use IterativelyIndexer abstract.

Options that won't be taken into account:
- since
- content-ids
- processes
- no-purge
EOT
            );
            $this->searchIndexer->createSearchIndex($output, (int) $iterationCount, !$commit);

            return 0;
        }

        $output->writeln('Re-indexing started for search engine: ' . $this->getName());
        $this->indexIteratively($input, $output, $iterationCount, $commit);
        $output->writeln('');
        $output->writeln('Finished re-indexing');
    }

    protected function indexIteratively(InputInterface $input, OutputInterface $output, $iterationCount, $commit)
    {
        if ($contentIds = $input->getOption('content-ids')) {
            $output->writeln('Indexing list of content id\'s');
            return $this->searchIndexer->updateSearchIndex($contentIds, $commit);
        }


        if ($since = $input->getOption('since')) {
            $stmt = $this->getStatementContentSince(new DateTime($since));
            $count = (int) $this->getStatementContentSince(new DateTime($since), 'count(c.id)')->fetchColumn();
            $purge = false;
        } else {
            $stmt = $this->getStatementContentAll();
            $count = (int) $this->getStatementContentAll('count(c.id)')->fetchColumn();
            $purge = !$input->getOption('no-purge');
        }

        if ($purge) {
            $output->writeln('Purging index before starting to re-create it');
            $this->searchIndexer->purge($commit);
            $output->writeln("Re-Creating Search Engine Index for {$count} content items..");
        } else {
            $output->writeln("Refresh Search Engine Index for {$count} content items..");
        }

        $iterations = ceil($count/$iterationCount);
        $progress = new ProgressBar($output);

        if (($processCount = (int) $input->getOption('processes')) > 1 && $processCount <= $iterations) {
            $output->writeln("Parallel indexing using $processCount processes sharing $iterations iterations");
            $progress->start($iterations);
            return $this->runParallelProcess($progress, $stmt, (int) $processCount, (int) $iterationCount, $commit);
        }

        // if we only have one process, or less iterations to warrant running several, we index it all inline
        $progress->start($iterations);
        foreach ($this->fetchIteration($stmt, $iterationCount) as $contentIds) {
            $this->searchIndexer->updateSearchIndex($contentIds, $commit);
            $progress->advance(1);
        }

        $progress->finish();
    }


    private function runParallelProcess(ProgressBar $progress, Statement $stmt, $processCount, $iterationCount, $commit)
    {
        /**
         * @var \Symfony\Component\Process\Process[]|null[] $processes
         */
        $processes = array_fill(0, $processCount, null);
        $generator = $this->fetchIteration($stmt, $iterationCount);
        do {
            foreach ($processes as $key => $process) {
                if ($process !== null && $process->isRunning()) {
                    continue;
                }

                if ($process !== null) {
                    // One of the processes just finished, so we increment progress bar
                    $progress->advance(1);
                }

                if (!$generator->valid()) {
                    unset($processes[$key]);
                    continue;
                }

                $processes[$key] = $this->getPhpProcess($generator->current(), $commit);
                $processes[$key]->start();
                $generator->next();
            }

            if (!empty($process)) {
                sleep(1);
            }

        } while (!empty($process));

        $progress->finish();
    }

    /**
     * @param DateTime $since
     * @param string $select Field to select, one of  'c.id' and 'count(c.id)'
     *
     * @return \Doctrine\DBAL\Driver\Statement
     */
    private function getStatementContentSince(DateTime $since, $select = 'c.id')
    {
        $q = $this->connection->createQueryBuilder()
            ->select($select)
            ->from('ezcontentobject', 'c')
            ->where('c.status = :status')->andWhere('c.modified >= :since')
            ->orderBy('c.modified', true)
            ->setParameter('status', ContentInfo::STATUS_PUBLISHED, PDO::PARAM_INT)
            ->setParameter('since', $since->getTimestamp(), PDO::PARAM_INT);

        return $q->execute();
    }

    /**
     * @param string $select Field to select, one of  'c.id' and 'count(c.id)'
     *
     * @return \Doctrine\DBAL\Driver\Statement
     */
    private function getStatementContentAll($select = 'c.id')
    {
        $q = $this->connection->createQueryBuilder()
            ->select($select)
            ->from('ezcontentobject', 'c')
            ->where('c.status = :status')
            ->setParameter('status', ContentInfo::STATUS_PUBLISHED, PDO::PARAM_INT);

        return $q->execute();
    }

    /**
     * @param \Doctrine\DBAL\Driver\Statement $stmt
     * @param int $iterationCount
     *
     * @return \Generator Return an array of arrays, each array contains content id's of $iterationCount.
     */
    private function fetchIteration(Statement $stmt, $iterationCount)
    {
        do {
            $contentIds = [];
            for ($i = 0; $i < $iterationCount; ++$i) {

This code seems to be duplicated across your project.

                if ($contentId = $stmt->fetch(PDO::FETCH_COLUMN)) {
                    $contentIds[] = $contentId;
                } else {
                    break;
                }
            }

            yield $contentIds;
        } while (!empty($contentId));
    }

    /**
     * @param array $contentIds
     * @param bool $commit
     *
     * @return \Symfony\Component\Process\Process
     */
    private static function getPhpProcess(array $contentIds, $commit)
    {
        $phpFinder = new PhpExecutableFinder();
        if (!$phpPath = $phpFinder->find()) {
            throw new \RuntimeException('The php executable could not be found, add it to your PATH environment variable and try again');
        }

        $php = escapeshellarg($phpPath);
        $console = escapeshellarg(file_exists('bin/console') ? 'bin/console' : 'app/console');
        $cmd = 'ezplatform:reindex --content-ids=' . implode(',', $contentIds). ($commit ? '' : ' --no-commit');

        return new Process($php.' '.$console.' '.$cmd, null, null, null, null);
    }

    /**
     * @return int
     */
    private function getNumberOfCPUCores()
    {
        $cores = 1;
        if (is_file('/proc/cpuinfo')) {
            // Linux (and potentially Windows with linux sub systems)
            $cpuinfo = file_get_contents('/proc/cpuinfo');
            preg_match_all('/^processor/m', $cpuinfo, $matches);
            $cores = count($matches[0]);
        } else if (DIRECTORY_SEPARATOR === '\\') {
            // Windows
            if (($process = @popen('wmic cpu get NumberOfCores', 'rb')) !== false) {
                fgets($process);
                $cores = (int) fgets($process);
                pclose($process);
            }
        } else if (($process = @popen('sysctl -a', 'rb')) !== false) {
            // *nix (Linux, BSD and Mac)
            $output = stream_get_contents($process);
            if (preg_match('/hw.ncpu: (\d+)/', $output, $matches)) {
                $cores = (int) $matches[1][0];
            }
            pclose($process);
        }

        return $cores;
    }
}