[TASK] Use bulk indexing in IndexingJob

This change register 100 node per IndexingJob to use bulk indexing and
fast ES index processing.

Author: dfeyer

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

@@ -107,21 +107,22 @@ public function updateAliasCommand($indexPostfix) {
 	 */
 	protected function indexWorkspace($workspaceName, $indexPostfix) {
 		$offset = 0;
-		$batchSize = 1000;
+		$batchSize = 100;
 		while (TRUE) {
-			$result = $this->nodeDataRepository->findAllBySiteAndWorkspace($workspaceName, $offset, 1000);
+			$result = $this->nodeDataRepository->findAllBySiteAndWorkspace($workspaceName, $offset, $batchSize);
 			if ($result === array()) {
 				break;
 			}
+			$jobData = [];
 			foreach ($result as $data) {
-				$indexingJob = new IndexingJob(
-					$indexPostfix,
-					$data['nodeIdentifier'],
-					$workspaceName,
-					$data['dimensions']
-				);
-				$this->jobManager->queue('Flowpack.ElasticSearch.ContentRepositoryQueueIndexer', $indexingJob);
+				$jobData[] = [
+					'nodeIdentifier' => $data['nodeIdentifier'],
+					'dimensions' => $data['dimensions']
+
+				];
 			}
+			$indexingJob = new IndexingJob($indexPostfix, $workspaceName, $jobData);
+			$this->jobManager->queue('Flowpack.ElasticSearch.ContentRepositoryQueueIndexer', $indexingJob);
 			$this->output('.');
 			$offset += $batchSize;
 		}

Classes/Flowpack/ElasticSearch/ContentRepositoryQueueIndexer/IndexingJob.php

@@ -14,6 +14,7 @@
 use Flowpack\ElasticSearch\ContentRepositoryAdaptor\Indexer\NodeIndexer;
 use Flowpack\ElasticSearch\ContentRepositoryAdaptor\LoggerInterface;
 use TYPO3\Flow\Annotations as Flow;
+use TYPO3\Flow\Utility\Algorithms;
 use TYPO3\Jobqueue\Common\Job\JobInterface;
 use TYPO3\Jobqueue\Common\Queue\Message;
 use TYPO3\Jobqueue\Common\Queue\QueueInterface;
@@ -29,66 +30,65 @@
 class IndexingJob implements JobInterface {
 
 	/**
-	 * @var string
+	 * @var NodeIndexer
+	 * @Flow\Inject
 	 */
-	protected $indexPostfix;
+	protected $nodeIndexer;
 
 	/**
-	 * @var string
+	 * @var NodeDataRepository
+	 * @Flow\Inject
 	 */
-	protected $nodeIdentifier;
+	protected $nodeDataRepository;
 
 	/**
-	 * @var string
+	 * @var NodeFactory
+	 * @Flow\Inject
 	 */
-	protected $workspaceName;
+	protected $nodeFactory;
 
 	/**
-	 * @var string
+	 * @var ContextFactory
+	 * @Flow\Inject
 	 */
-	protected $dimensions;
+	protected $contextFactory;
 
 	/**
+	 * @var LoggerInterface
 	 * @Flow\Inject
-	 * @var NodeIndexer
 	 */
-	protected $nodeIndexer;
+	protected $logger;
 
 	/**
-	 * @Flow\Inject
-	 * @var NodeDataRepository
+	 * @var string
 	 */
-	protected $nodeDataRepository;
+	protected $identifier;
 
 	/**
-	 * @Flow\Inject
-	 * @var NodeFactory
+	 * @var string
 	 */
-	protected $nodeFactory;
+	protected $workspaceName;
 
 	/**
-	 * @Flow\Inject
-	 * @var ContextFactory
+	 * @var string
 	 */
-	protected $contextFactory;
+	protected $indexPostfix;
 
 	/**
-	 * @Flow\Inject
-	 * @var LoggerInterface
+	 * @var array
 	 */
-	protected $logger;
+	protected $nodes = [];
 
 	/**
 	 * @param string $indexPostfix
-	 * @param string $nodeIdentifier
 	 * @param string $workspaceName
-	 * @param array $dimensions
+	 * @param array $nodes
 	 */
-	public function __construct($indexPostfix, $nodeIdentifier, $workspaceName, array $dimensions = array()) {
-		$this->indexPostfix = $indexPostfix;
-		$this->nodeIdentifier = $nodeIdentifier;
+	public function __construct($indexPostfix, $workspaceName, array $nodes) {
+		$this->identifier = Algorithms::generateRandomString(24);
 		$this->workspaceName = $workspaceName;
-		$this->dimensions = $dimensions;
+		$this->indexPostfix = $indexPostfix;
+		$this->nodes = $nodes;
 	}
 
 	/**
@@ -100,18 +100,21 @@ public function __construct($indexPostfix, $nodeIdentifier, $workspaceName, arra
 	 * @return boolean TRUE if the job was executed successfully and the message should be finished
 	 */
 	public function execute(QueueInterface $queue, Message $message) {
-		/** @var NodeData $nodeData */
-		$nodeData = $this->nodeDataRepository->findByIdentifier($this->nodeIdentifier);
-		$context = $this->contextFactory->create([
-			'workspaceName' => $this->workspaceName,
-			'dimensions' => $this->dimensions
-		]);
-		$currentNode = $this->nodeFactory->createFromNodeData($nodeData, $context);
-		if (!$currentNode instanceof NodeInterface) {
-			return TRUE;
+		foreach ($this->nodes as $node) {
+			/** @var NodeData $nodeData */
+			$nodeData = $this->nodeDataRepository->findByIdentifier($node['nodeIdentifier']);
+			$context = $this->contextFactory->create([
+				'workspaceName' => $this->workspaceName,
+				'dimensions' => $node['dimensions']
+			]);
+			$currentNode = $this->nodeFactory->createFromNodeData($nodeData, $context);
+			if (!$currentNode instanceof NodeInterface) {
+				return TRUE;
+			}
+			$this->nodeIndexer->setIndexNamePostfix($this->indexPostfix);
+			$this->nodeIndexer->indexNode($currentNode);
 		}
-		$this->nodeIndexer->setIndexNamePostfix($this->indexPostfix);
-		$this->nodeIndexer->indexNode($currentNode);
+
 		$this->nodeIndexer->flush();
 
 		return TRUE;
@@ -123,7 +126,7 @@ public function execute(QueueInterface $queue, Message $message) {
 	 * @return string A job identifier
 	 */
 	public function getIdentifier() {
-		return $this->nodeIdentifier;
+		return $this->identifier;
 	}
 
 	/**