Merge branch 'master' into johannessteu-master

# Conflicts:
#	Classes/Flowpack/ElasticSearch/ContentRepositoryQueueIndexer/Command/NodeIndexQueueCommandController.php
#	Classes/Flowpack/ElasticSearch/ContentRepositoryQueueIndexer/IndexingJob.php
#	Classes/Flowpack/ElasticSearch/ContentRepositoryQueueIndexer/UpdateAliasJob.php
#	Configuration/Settings.yaml
#	composer.json

Author: dfeyer

Classes/Flowpack/ElasticSearch/ContentRepositoryQueueIndexer/Command/NodeIndexQueueCommandController.php

@@ -6,6 +6,7 @@
 use Flowpack\ElasticSearch\ContentRepositoryAdaptor\Mapping\NodeTypeMappingBuilder;
 use Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\Domain\Repository\NodeDataRepository;
 use Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\IndexingJob;
+use Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\LoggerTrait;
 use Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\UpdateAliasJob;
 use Flowpack\JobQueue\Common\Job\JobManager;
 use Neos\Flow\Annotations as Flow;
@@ -20,47 +21,52 @@
  */
 class NodeIndexQueueCommandController extends CommandController
 {
+    use LoggerTrait;
+
+    const BATCH_QUEUE_NAME = 'Flowpack.ElasticSearch.ContentRepositoryQueueIndexer';
+    const LIVE_QUEUE_NAME = 'Flowpack.ElasticSearch.ContentRepositoryQueueIndexer.Live';
+
     /**
-     * @Flow\Inject
      * @var JobManager
+     * @Flow\Inject
      */
     protected $jobManager;
 
+    /**
+     * @var QueueManager
+     * @Flow\Inject
+     */
+    protected $queueManager;
+
     /**
      * @var PersistenceManagerInterface
      * @Flow\Inject
      */
     protected $persistenceManager;
 
     /**
-     * @Flow\Inject
      * @var NodeTypeMappingBuilder
+     * @Flow\Inject
      */
     protected $nodeTypeMappingBuilder;
 
     /**
-     * @Flow\Inject
      * @var NodeDataRepository
+     * @Flow\Inject
      */
     protected $nodeDataRepository;
 
     /**
-     * @Flow\Inject
      * @var WorkspaceRepository
+     * @Flow\Inject
      */
     protected $workspaceRepository;
 
     /**
-     * @Flow\Inject
      * @var NodeIndexer
-     */
-    protected $nodeIndexer;
-
-    /**
      * @Flow\Inject
-     * @var LoggerInterface
      */
-    protected $logger;
+    protected $nodeIndexer;
 
     /**
      * Index all nodes by creating a new index and when everything was completed, switch the index alias.
@@ -70,29 +76,128 @@ class NodeIndexQueueCommandController extends CommandController
     public function buildCommand($workspace = null)
     {
         $indexPostfix = time();
-        $this->createNextIndex($indexPostfix);
+        $indexName = $this->createNextIndex($indexPostfix);
         $this->updateMapping();
 
-        $this->outputLine(sprintf('Indexing on %s ... ', $indexPostfix));
+
+        $this->outputLine();
+        $this->outputLine('<b>Indexing on %s ...</b>', [$indexName]);
+
+        $pendingJobs = $this->queueManager->getQueue(self::BATCH_QUEUE_NAME)->count();
+        if ($pendingJobs !== 0) {
+            $this->outputLine('<error>!! </error> The queue "%s" is not empty (%d pending jobs), please flush the queue.', [self::BATCH_QUEUE_NAME, $pendingJobs]);
+            $this->quit(1);
+        }
 
         if ($workspace === null) {
             foreach ($this->workspaceRepository->findAll() as $workspace) {
                 $workspace = $workspace->getName();
                 $this->outputLine();
-                $this->outputLine(sprintf('<info>++</info> Indexing %s workspace', $workspace));
                 $this->indexWorkspace($workspace, $indexPostfix);
             }
         } else {
             $this->outputLine();
-            $this->outputLine(sprintf('<info>++</info> Indexing only %s workspace', $workspace));
             $this->indexWorkspace($workspace, $indexPostfix);
         }
         $updateAliasJob = new UpdateAliasJob($indexPostfix);
-        $queueName = 'Flowpack.ElasticSearch.ContentRepositoryQueueIndexer';
-        $this->jobManager->queue($queueName, $updateAliasJob);
+        $this->jobManager->queue(self::BATCH_QUEUE_NAME, $updateAliasJob);
+
+        $this->outputLine("Indexing jobs created for queue %s with success ...", [self::BATCH_QUEUE_NAME]);
+        $this->outputSystemReport();
+        $this->outputLine();
+    }
 
+    /**
+     * @param string $queue Type of queue to process, can be "live" or "batch"
+     * @param int $exitAfter If set, this command will exit after the given amount of seconds
+     * @param int $limit If set, only the given amount of jobs are processed (successful or not) before the script exits
+     * @param bool $verbose Output debugging information
+     * @return void
+     */
+    public function workCommand($queue = 'batch', $exitAfter = null, $limit = null, $verbose = false)
+    {
+        $allowedQueues = [
+            'batch' => self::BATCH_QUEUE_NAME,
+            'live' => self::LIVE_QUEUE_NAME
+        ];
+        if (!isset($allowedQueues[$queue])) {
+            $this->output('Invalid queue, should be "live" or "batch"');
+        }
+        $queueName = $allowedQueues[$queue];
+
+        if ($verbose) {
+            $this->output('Watching queue <b>"%s"</b>', [$queueName]);
+            if ($exitAfter !== null) {
+                $this->output(' for <b>%d</b> seconds', [$exitAfter]);
+            }
+            $this->outputLine('...');
+        }
+
+        $startTime = time();
+        $timeout = null;
+        $numberOfJobExecutions = 0;
+
+        do {
+            $message = null;
+            if ($exitAfter !== null) {
+                $timeout = max(1, $exitAfter - (time() - $startTime));
+            }
+            try {
+                $message = $this->jobManager->waitAndExecute($queueName, $timeout);
+            } catch (JobQueueException $exception) {
+                $numberOfJobExecutions ++;
+                $this->outputLine('<error>%s</error>', [$exception->getMessage()]);
+                if ($verbose && $exception->getPrevious() instanceof \Exception) {
+                    $this->outputLine('  Reason: %s', [$exception->getPrevious()->getMessage()]);
+                }
+            } catch (\Exception $exception) {
+                $this->outputLine('<error>Unexpected exception during job execution: %s, aborting...</error>', [$exception->getMessage()]);
+                $this->quit(1);
+            }
+            if ($message !== null) {
+                $numberOfJobExecutions ++;
+                if ($verbose) {
+                    $messagePayload = strlen($message->getPayload()) <= 50 ? $message->getPayload() : substr($message->getPayload(), 0, 50) . '...';
+                    $this->outputLine('<success>Successfully executed job "%s" (%s)</success>', [$message->getIdentifier(), $messagePayload]);
+                }
+            }
+            if ($exitAfter !== null && (time() - $startTime) >= $exitAfter) {
+                if ($verbose) {
+                    $this->outputLine('Quitting after %d seconds due to <i>--exit-after</i> flag', [time() - $startTime]);
+                }
+                $this->quit();
+            }
+            if ($limit !== null && $numberOfJobExecutions >= $limit) {
+                if ($verbose) {
+                    $this->outputLine('Quitting after %d executed job%s due to <i>--limit</i> flag', [$numberOfJobExecutions, $numberOfJobExecutions > 1 ? 's' : '']);
+                }
+                $this->quit();
+            }
+
+        } while (true);
+    }
+
+    /**
+     * Flush the index queue
+     */
+    public function flushCommand()
+    {
+        $this->queueManager->getQueue(self::BATCH_QUEUE_NAME)->flush();
+        $this->outputSystemReport();
         $this->outputLine();
-        $this->outputLine(sprintf('Indexing jobs created for queue %s with success ...', $queueName));
+    }
+
+    /**
+     * Output system report for CLI commands
+     */
+    protected function outputSystemReport()
+    {
+        $this->outputLine();
+        $this->outputLine('Memory Usage   : %s', [Files::bytesToSizeString(memory_get_peak_usage(true))]);
+        $time = microtime(true) - $_SERVER["REQUEST_TIME_FLOAT"];
+        $this->outputLine('Execution time : %s seconds', [$time]);
+        $this->outputLine('Indexing Queue : %s', [self::BATCH_QUEUE_NAME]);
+        $this->outputLine('Pending Jobs   : %s', [$this->queueManager->getQueue(self::BATCH_QUEUE_NAME)->count()]);
     }
 
     /**
@@ -101,6 +206,8 @@ public function buildCommand($workspace = null)
      */
     protected function indexWorkspace($workspaceName, $indexPostfix)
     {
+        $this->outputLine('<info>++</info> Indexing %s workspace', [$workspaceName]);
+        $nodeCounter = 0;
         $offset = 0;
         $batchSize = 500;
         while (true) {
@@ -113,30 +220,34 @@ protected function indexWorkspace($workspaceName, $indexPostfix)
                     'nodeIdentifier' => $data['nodeIdentifier'],
                     'dimensions' => $data['dimensions']
                 ];
+                $nodeCounter++;
             }
 
             if ($jobData === []) {
                 break;
             }
 
             $indexingJob = new IndexingJob($indexPostfix, $workspaceName, $jobData);
-            $this->jobManager->queue('Flowpack.ElasticSearch.ContentRepositoryQueueIndexer', $indexingJob);
+            $this->jobManager->queue(self::BATCH_QUEUE_NAME, $indexingJob);
             $this->output('.');
             $offset += $batchSize;
             $this->persistenceManager->clearState();
         }
         $this->outputLine();
+        $this->outputLine("\nNumber of Nodes to be indexed in workspace '%s': %d", [$workspaceName, $nodeCounter]);
+        $this->outputLine();
     }
 
     /**
-     * Create next index
      * @param string $indexPostfix
+     * @return string
      */
     protected function createNextIndex($indexPostfix)
     {
         $this->nodeIndexer->setIndexNamePostfix($indexPostfix);
         $this->nodeIndexer->getIndex()->create();
-        $this->logger->log(sprintf('action=indexing step=index-created index=%s', $this->nodeIndexer->getIndexName()), LOG_INFO);
+        $this->log(sprintf('action=indexing step=index-created index=%s', $this->nodeIndexer->getIndexName()), LOG_INFO);
+        return $this->nodeIndexer->getIndexName();
     }
 
     /**
@@ -146,9 +257,9 @@ protected function updateMapping()
     {
         $nodeTypeMappingCollection = $this->nodeTypeMappingBuilder->buildMappingInformation($this->nodeIndexer->getIndex());
         foreach ($nodeTypeMappingCollection as $mapping) {
-            /** @var \Flowpack\ElasticSearch\Domain\Model\Mapping $mapping */
+            /** @var Mapping $mapping */
             $mapping->apply();
         }
-        $this->logger->log(sprintf('action=indexing step=mapping-updated index=%s', $this->nodeIndexer->getIndexName()), LOG_INFO);
+        $this->log(sprintf('action=indexing step=mapping-updated index=%s', $this->nodeIndexer->getIndexName()), LOG_INFO);
     }
 }

Classes/Flowpack/ElasticSearch/ContentRepositoryQueueIndexer/Indexer/NodeIndexer.php

@@ -0,0 +1,53 @@
+<?php
+namespace Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\Indexer;
+
+use Flowpack\ElasticSearch\ContentRepositoryAdaptor;
+use Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\Command\NodeIndexQueueCommandController;
+use Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\IndexingJob;
+use Flowpack\JobQueue\Common\Job\JobManager;
+use TYPO3\Flow\Annotations as Flow;
+use TYPO3\Flow\Persistence\PersistenceManagerInterface;
+use TYPO3\TYPO3CR\Domain\Model\NodeInterface;
+
+/**
+ * ElasticSearch Indexing Job Interface
+ */
+class NodeIndexer extends ContentRepositoryAdaptor\Indexer\NodeIndexer
+{
+    /**
+     * @var JobManager
+     * @Flow\Inject
+     */
+    protected $jobManager;
+
+    /**
+     * @var PersistenceManagerInterface
+     * @Flow\Inject
+     */
+    protected $persistenceManager;
+
+    /**
+     * @var bool
+     * @Flow\InjectConfiguration(path="enableLiveAsyncIndexing")
+     */
+    protected $enableLiveAsyncIndexing;
+
+    /**
+     * @param NodeInterface $node
+     * @param string|null $targetWorkspaceName
+     */
+    public function indexNode(NodeInterface $node, $targetWorkspaceName = null)
+    {
+        if ($this->enableLiveAsyncIndexing !== true) {
+            parent::indexNode($node, $targetWorkspaceName);
+            return;
+        }
+        $indexingJob = new IndexingJob($this->indexNamePostfix, $targetWorkspaceName, [
+            [
+                'nodeIdentifier' => $this->persistenceManager->getIdentifierByObject($node->getNodeData()),
+                'dimensions' => $node->getDimensions()
+            ]
+        ]);
+        $this->jobManager->queue(NodeIndexQueueCommandController::LIVE_QUEUE_NAME, $indexingJob);
+    }
+}

Classes/Flowpack/ElasticSearch/ContentRepositoryQueueIndexer/IndexingJob.php

@@ -18,6 +18,8 @@
  */
 class IndexingJob implements JobInterface
 {
+    use LoggerTrait;
+
     /**
      * @var NodeIndexer
      * @Flow\Inject
@@ -42,12 +44,6 @@ class IndexingJob implements JobInterface
      */
     protected $contextFactory;
 
-    /**
-     * @var SystemLoggerInterface
-     * @Flow\Inject
-     */
-    protected $logger;
-
     /**
      * @var string
      */
@@ -107,20 +103,20 @@ public function execute(QueueInterface $queue, Message $message)
 
                 // Skip this iteration if the node can not be fetched from the current context
                 if (!$currentNode instanceof NodeInterface) {
-                    $this->logger->log(sprintf('action=indexing step=failed node=%s message="Node could not be processed"', $node['nodeIdentifier']));
+                    $this->log(sprintf('action=indexing step=failed node=%s message="Node could not be processed"', $node['nodeIdentifier']));
                     continue;
                 }
 
                 $this->nodeIndexer->setIndexNamePostfix($this->indexPostfix);
-                $this->logger->log(sprintf('action=indexing step=started node=%s', $currentNode->getIdentifier()));
+                $this->log(sprintf('action=indexing step=started node=%s', $currentNode->getIdentifier()));
 
                 $this->nodeIndexer->indexNode($currentNode);
             }
 
             $this->nodeIndexer->flush();
             $duration = microtime(true) - $startTime;
             $rate = $numberOfNodes / $duration;
-            $this->logger->log(sprintf('action=indexing step=finished number_of_nodes=%d duration=%f nodes_per_second=%f', $numberOfNodes, $duration, $rate));
+            $this->log(sprintf('action=indexing step=finished number_of_nodes=%d duration=%f nodes_per_second=%f', $numberOfNodes, $duration, $rate));
         });
 
         return true;

Classes/Flowpack/ElasticSearch/ContentRepositoryQueueIndexer/LoggerTrait.php

@@ -0,0 +1,32 @@
+<?php
+namespace Flowpack\ElasticSearch\ContentRepositoryQueueIndexer;
+
+use TYPO3\Flow\Annotations as Flow;
+
+/**
+ * LoggerTrait
+ */
+trait LoggerTrait
+{
+    /**
+     * @var \TYPO3\Flow\Log\SystemLoggerInterface
+     * @Flow\Inject
+     */
+    protected $_logger;
+
+    /**
+     * @param string $message The message to log
+     * @param integer $severity An integer value, one of the LOG_* constants
+     * @param mixed $additionalData A variable containing more information about the event to be logged
+     * @param string $packageKey Key of the package triggering the log (determined automatically if not specified)
+     * @param string $className Name of the class triggering the log (determined automatically if not specified)
+     * @param string $methodName Name of the method triggering the log (determined automatically if not specified)
+     * @return void
+     * @api
+     */
+    protected function log($message, $severity = LOG_INFO, $additionalData = null, $packageKey = null, $className = null, $methodName = null)
+    {
+        $packageKey = $packageKey ?: 'Flowpack.ElasticSearch.ContentRepositoryQueueIndexer';
+        $this->_logger->log($message, $severity, $additionalData, $packageKey, $className, $methodName);
+    }
+}