0.7

.travis.yml

@@ -0,0 +1,7 @@
+install:
+- curl -sS https://getcomposer.org/installer | php
+before_script:
+- php composer.phar update -n
+script:
+- vendor/bin/phpcs . --ignore=*/vendor/* --standard=PSR2 --extensions=php
+- vendor/bin/phpunit -c phpunit.xml

Command/IndexerQueueCommand.php

@@ -14,11 +14,12 @@
 use DateTime;
 use DateTimeZone;
 
-use Integrated\Common\Content\ContentInterface;
 use Integrated\Common\ContentType\ResolverInterface;
 use Integrated\Common\Solr\Indexer\Job;
 use Integrated\Common\Queue\QueueInterface;
 
+use Integrated\Bundle\ContentBundle\Document\Content\Content;
+
 use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand;
 
 use Symfony\Component\Console\Helper\ProgressHelper;
@@ -175,32 +176,21 @@ protected function executeDelete(InputInterface $input, OutputInterface $output)
      */
     protected function executeIndex(InputInterface $input, OutputInterface $output)
     {
-        $result = null;
+        // Don't hydrate for performance reasons
+        $builder = $this->getDocumentManager()->createQueryBuilder(Content::class);
+        $builder->select('id', 'contentType', 'class')->hydrate(false);
 
         if ($input->getOption('full')) {
-
-            //use createQueryBuilder for performance reasons
-            $qb = $this->getDocumentManager()->createQueryBuilder(
-                'Integrated\Bundle\ContentBundle\Document\Content\Content'
-            )
-                ->select('id', 'contentType', 'class');
-            $result = $qb->getQuery()->execute();
+            $result = $builder->getQuery()->execute();
 
             // The entire site is going to be reindex so everything that is now in the queue
             // will be redone so just clear it so content is not double indexed.
 
             $this->getQueue()->clear();
         } else {
-            $criteria['$or'] = [];
+            $builder->field('contentType')->in($input->getArgument('id'));
 
-            foreach ($input->getArgument('id') as $id) {
-                $criteria['$or'][] = ['contentType' => $id];
-            }
-
-            $result = $this->getDocumentManager()
-                ->getUnitOfWork()
-                ->getDocumentPersister('Integrated\Bundle\ContentBundle\Document\Content\Content')
-                ->loadAll($criteria);
+            $result = $builder->getQuery()->execute();
         }
 
         if ($count = $result->count()) {
@@ -251,17 +241,17 @@ protected function doIndex(Cursor $cursor, ProgressHelper $progress)
         $count = 0;
         $manager = $this->getDocumentManager();
 
-        /** @var ContentInterface $document */
-
         foreach ($cursor as $document) {
             $progress->advance();
 
             $job = new Job('ADD');
 
-            $job->setOption('document.id', $document->getContentType() . '-' . $document->getId());
+            $contentType = isset($document['contentType']) ? $document['contentType'] : '';
+
+            $job->setOption('document.id', $contentType . '-' . $document['_id']);
 
-            $job->setOption('document.data', $this->getSerializer()->serialize($document, 'json'));
-            $job->setOption('document.class', get_class($document));
+            $job->setOption('document.data', json_encode(['id' => $document['_id']]));
+            $job->setOption('document.class', $document['class']);
             $job->setOption('document.format', 'json');
 
             $queue->push($job);

Command/IndexerRunCommand.php

@@ -11,22 +11,72 @@
 
 namespace Integrated\Bundle\SolrBundle\Command;
 
-use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand;
+use Exception;
+
+use Integrated\Bundle\SolrBundle\EventListener\DoctrineClearEventSubscriber;
+use Integrated\Bundle\SolrBundle\Process\ArgumentProcess;
+use Integrated\Bundle\SolrBundle\Process\ProcessPoolGenerator;
+
+use Integrated\Common\Queue\Provider\DBAL\QueueProvider;
+use Integrated\Common\Solr\Indexer\Indexer;
+
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\HttpKernel\KernelInterface;
 use Symfony\Component\Process\Process;
 use Symfony\Component\Filesystem\LockHandler;
 
-use Exception;
-
-use Integrated\Common\Solr\Indexer\IndexerInterface;
-
 /**
  * @author Jan Sanne Mulder <[email protected]>
  */
-class IndexerRunCommand extends ContainerAwareCommand
+class IndexerRunCommand extends Command
 {
+    /**
+     * @var Indexer
+     */
+    protected $indexer;
+
+    /**
+     * @var QueueProvider
+     */
+    protected $queueProvider;
+
+    /**
+     * @var string
+     */
+    protected $workingDirectory;
+
+    /**
+     * @var DoctrineClearEventSubscriber
+     */
+    protected $clearEventSubscriber;
+
+    /**
+     * @var KernelInterface
+     */
+    protected $kernel;
+
+    /**
+     * @param Indexer $indexer
+     * @param QueueProvider $queueProvider
+     * @param KernelInterface $kernel
+     * @param DoctrineClearEventSubscriber $clearEventSubscriber
+     * @param string $workingDirectory
+     */
+    public function __construct(Indexer $indexer, QueueProvider $queueProvider, DoctrineClearEventSubscriber $clearEventSubscriber, KernelInterface $kernel, $workingDirectory)
+    {
+        parent::__construct();
+
+        $this->indexer = $indexer;
+        $this->queueProvider = $queueProvider;
+        $this->workingDirectory = $workingDirectory;
+        $this->clearEventSubscriber = $clearEventSubscriber;
+        $this->kernel = $kernel;
+    }
+
     /**
      * @see Command
      */
@@ -48,7 +98,19 @@ protected function configure()
                 'Time in milliseconds to wait between runs (in combination with --full or --daemon)',
                 0
             )
-            ->setDescription('Execute a sol indexer run')
+            ->addArgument(
+                'processes',
+                InputArgument::OPTIONAL,
+                'Creates a number of proccess that run the queue',
+                0
+            )
+            ->addOption(
+                'blocking',
+                'b',
+                InputOption::VALUE_NONE,
+                'Block the current command until all sub-processes are done'
+            )
+            ->setDescription('Execute a solr indexer run')
             ->setHelp('
 The <info>%command.name%</info> command starts a indexer run.
 
@@ -58,39 +120,44 @@ protected function configure()
     }
 
     /**
-     * @see Command::execute()
+     * @param InputInterface $input
+     * @param OutputInterface $output
+     * @return int
      */
     protected function execute(InputInterface $input, OutputInterface $output)
     {
-        if ($input->getOption('full') || $input->getOption('daemon')) {
+        if ($argument = $input->getArgument('processes')) {
+            return $this->runProcess(new ArgumentProcess($argument), $input, $output);
+        } else if ($input->getOption('full') || $input->getOption('daemon')) {
             return $this->runExternal($input, $output);
         }
 
-        return $this->runInternal($input, $output);
+        return $this->runInternal(self::class, $output);
     }
 
     /**
-     * @param InputInterface $input
+     * @param string $lock
      * @param OutputInterface $output
      * @return int
      */
-    private function runInternal(InputInterface $input, OutputInterface $output)
+    private function runInternal($lock, OutputInterface $output)
     {
         try {
-            $lock = new LockHandler('Integrated\Bundle\SolrBundle\Command\IndexerRunCommand');
-            $attemps = 0;
+            $lock = new LockHandler($lock);
+            $attempts = 0;
             while (!$lock->lock()) {
-                //retry for almost a minute, otherwise don't throw an error (after all another indexer is running)
-                if ($attemps++ >= 10) {
+                // Retry for almost a minute, otherwise don't throw an error (after all another indexer is running)
+                if ($attempts++ >= 10) {
                     return 0;
                 }
                 sleep(5);
             }
 
-            /** @var IndexerInterface $indexer */
-            $indexer = $this->getContainer()->get('integrated_solr.indexer');
-            $indexer->execute();
+            if ($output->isDebug() && method_exists($this->indexer, 'setDebug')) {
+                $this->indexer->setDebug();
+            }
 
+            $this->indexer->execute();
         } catch (Exception $e) {
             $output->writeln("Aborting: " . $e->getMessage());
 
@@ -107,24 +174,27 @@ private function runInternal(InputInterface $input, OutputInterface $output)
      */
     private function runExternal(InputInterface $input, OutputInterface $output)
     {
-        $wait = (int)$input->getOption('wait');
+        $wait = (int) $input->getOption('wait');
         $wait = $wait * 1000; // convert from milli to micro
 
         while (true) {
+            // Run a external process
             $process = new Process(
-                'php app/console solr:indexer:run -e ' . $input->getOption('env'),
-                $this->getRootDir()
+                sprintf('php app/console solr:indexer:run -e %s', $this->kernel->getEnvironment()),
+                $this->workingDirectory
             );
 
             $process->setTimeout(0);
-            $process->run();
+            $process->run(function ($type, $buffer) use ($output) {
+                $output->write($buffer, false, $type);
+            });
 
             if (!$process->isSuccessful()) {
                 break; // terminate when there is a error
             }
 
             if (!$input->getOption('daemon')) {
-                if (!$this->getContainer()->get('integrated_solr.indexer')->getQueue()->count()) {
+                if (!$this->indexer->getQueue()->count()) {
                     break;
                 }
             }
@@ -136,10 +206,72 @@ private function runExternal(InputInterface $input, OutputInterface $output)
     }
 
     /**
-     * @return string
+     * @param ArgumentProcess $argument
+     * @param InputInterface $input
+     * @param OutputInterface $output
+     * @return int
      */
-    protected function getRootDir()
+    private function runProcess(ArgumentProcess $argument, InputInterface $input, OutputInterface $output)
     {
-        return realpath($this->getContainer()->get('kernel')->getRootDir() . '/..');
+        if ($argument->isParentProcess()) {
+            // Create pool generator to generate the processes
+            $generator = new ProcessPoolGenerator($input, $this->kernel);
+            $pool = $generator->getProcessesPool($argument, $this->workingDirectory);
+
+            // Start them accordingly
+            foreach ($pool as $i => $process) {
+                // Run it
+                $process->start();
+
+                // Tell somebody
+                $output->writeln(sprintf('Started process %d with pid %d to run the queue', ($i+1), $process->getPid()));
+            }
+
+            if ($input->getOption('blocking')) {
+                $output->writeln('Running in blocking mode, waiting until all started processes are done');
+
+                // While the pool contains processes we're running
+                while ($pool->count()) {
+                    foreach ($pool as $i => $process) {
+                        // Read stout for anything to pass thru
+                        if ($processOutput = $process->getIncrementalOutput()) {
+                            $output->writeln(sprintf('Prcocess %d: %s', $i, $processOutput));
+                        }
+                        // Read sterr for anything to pass thru
+                        if ($processOutput = $process->getIncrementalErrorOutput()) {
+                            $output->writeln(sprintf('Prcocess %d: %s', $i, $processOutput));
+                        }
+
+                        if (!$process->isRunning()) {
+                            // Tell the user
+                            $output->writeln(sprintf('Process %d finished', ($i+1)));
+
+                            // This one is important
+                            $pool->removeElement($process);
+                        }
+                    }
+
+                    // Don't create a cpu load, check periodically
+                    sleep(1);
+                }
+            }
+        } else {
+            // Set the modulo to run over the data set with x processes, creating a unique list per thread
+            $this->queueProvider->setOption('where', sprintf('(id %% %d) = %d', $argument->getProcessMax(), $argument->getProcessNumber()));
+
+            // Add the clear event listener only for the thread
+            $this->indexer->getEventDispatcher()->addSubscriber($this->clearEventSubscriber);
+
+            // Seems to be a sub-process, ran it with a the number appended to the class
+            while ($this->indexer->getQueue()->count()) {
+                $this->runInternal(sprintf('%s:%d', self::class, $argument->getProcessNumber()), $output);
+
+                // Give them cores some relaxation
+                usleep(5000);
+            }
+        }
+
+        // Good to go
+        return 0;
     }
 }