Merge pull request #39 from daniellienert/task/cradapter-v6-compat

WIP: TASK: Refactor package / make compatible to ES CRAdapter 6.x

Author: daniellienert

Classes/AbstractIndexingJob.php

@@ -19,10 +19,16 @@
 use Neos\ContentRepository\Domain\Service\ContextFactoryInterface;
 use Neos\Flow\Annotations as Flow;
 use Neos\Flow\Utility\Algorithms;
+use Psr\Log\LoggerInterface;
 
 abstract class AbstractIndexingJob implements JobInterface
 {
-    use LoggerTrait;
+
+    /**
+     * @FLow\Inject
+     * @var LoggerInterface
+     */
+    protected $logger;
 
     /**
      * @var NodeIndexer
@@ -78,6 +84,7 @@ abstract class AbstractIndexingJob implements JobInterface
      * @param string $indexPostfix
      * @param string $targetWorkspaceName In case indexing is triggered during publishing, a target workspace name will be passed in
      * @param array $nodes
+     * @throws \Exception
      */
     public function __construct($indexPostfix, $targetWorkspaceName, array $nodes)
     {
@@ -92,7 +99,7 @@ public function __construct($indexPostfix, $targetWorkspaceName, array $nodes)
      *
      * @return string A job identifier
      */
-    public function getIdentifier()
+    public function getIdentifier(): string
     {
         return $this->identifier;
     }

Classes/Command/NodeIndexQueueCommandController.php

@@ -1,4 +1,6 @@
 <?php
+declare(strict_types=1);
+
 namespace Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\Command;
 
 /*
@@ -12,6 +14,7 @@
  */
 
 use Flowpack\ElasticSearch\ContentRepositoryAdaptor\Driver\NodeTypeMappingBuilderInterface;
+use Flowpack\ElasticSearch\ContentRepositoryAdaptor\Exception\ConfigurationException;
 use Flowpack\ElasticSearch\ContentRepositoryAdaptor\Indexer\NodeIndexer;
 use Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\Domain\Repository\NodeDataRepository;
 use Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\IndexingJob;
@@ -24,8 +27,12 @@
 use Neos\ContentRepository\Domain\Repository\WorkspaceRepository;
 use Neos\Flow\Annotations as Flow;
 use Neos\Flow\Cli\CommandController;
+use Neos\Flow\Cli\Exception\StopCommandException;
+use Neos\Flow\Log\Utility\LogEnvironment;
+use Neos\Flow\Mvc\Exception\StopActionException;
 use Neos\Flow\Persistence\PersistenceManagerInterface;
 use Neos\Utility\Files;
+use Psr\Log\LoggerInterface;
 
 /**
  * Provides CLI features for index handling
@@ -34,10 +41,14 @@
  */
 class NodeIndexQueueCommandController extends CommandController
 {
-    use LoggerTrait;
+    protected const BATCH_QUEUE_NAME = 'Flowpack.ElasticSearch.ContentRepositoryQueueIndexer';
+    public const LIVE_QUEUE_NAME = 'Flowpack.ElasticSearch.ContentRepositoryQueueIndexer.Live';
 
-    const BATCH_QUEUE_NAME = 'Flowpack.ElasticSearch.ContentRepositoryQueueIndexer';
-    const LIVE_QUEUE_NAME = 'Flowpack.ElasticSearch.ContentRepositoryQueueIndexer.Live';
+    /**
+     * @Flow\Inject
+     * @var LoggerInterface
+     */
+    protected $logger;
 
     /**
      * @var JobManager
@@ -91,13 +102,17 @@ class NodeIndexQueueCommandController extends CommandController
      * Index all nodes by creating a new index and when everything was completed, switch the index alias.
      *
      * @param string $workspace
-     * @throws \Flowpack\JobQueue\Common\Exception
-     * @throws \Neos\Flow\Mvc\Exception\StopActionException
+     * @throws ConfigurationException
+     * @throws Exception
+     * @throws StopCommandException
      * @throws \Flowpack\ElasticSearch\ContentRepositoryAdaptor\Exception
+     * @throws \Flowpack\ElasticSearch\Exception
+     * @throws \Neos\Flow\Http\Exception
+     * @throws \Exception
      */
-    public function buildCommand($workspace = null)
+    public function buildCommand(string $workspace = null): void
     {
-        $indexPostfix = time();
+        $indexPostfix = (string) time();
         $indexName = $this->createNextIndex($indexPostfix);
         $this->updateMapping($indexPostfix);
 
@@ -120,10 +135,11 @@ public function buildCommand($workspace = null)
             $this->outputLine();
             $this->indexWorkspace($workspace, $indexPostfix);
         }
+
         $updateAliasJob = new UpdateAliasJob($indexPostfix);
         $this->jobManager->queue(self::BATCH_QUEUE_NAME, $updateAliasJob);
 
-        $this->outputLine("Indexing jobs created for queue %s with success ...", [self::BATCH_QUEUE_NAME]);
+        $this->outputLine('Indexing jobs created for queue %s with success ...', [self::BATCH_QUEUE_NAME]);
         $this->outputSystemReport();
         $this->outputLine();
     }
@@ -134,9 +150,10 @@ public function buildCommand($workspace = null)
      * @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
-     * @throws \Neos\Flow\Mvc\Exception\StopActionException
+     * @throws StopActionException
+     * @throws StopCommandException
      */
-    public function workCommand($queue = 'batch', $exitAfter = null, $limit = null, $verbose = false)
+    public function workCommand(string $queue = 'batch', int $exitAfter = null, int $limit = null, $verbose = false): void
     {
         $allowedQueues = [
             'batch' => self::BATCH_QUEUE_NAME,
@@ -201,7 +218,7 @@ public function workCommand($queue = 'batch', $exitAfter = null, $limit = null,
     /**
      * Flush the index queue
      */
-    public function flushCommand()
+    public function flushCommand(): void
     {
         try {
             $this->queueManager->getQueue(self::BATCH_QUEUE_NAME)->flush();
@@ -235,8 +252,9 @@ protected function outputSystemReport()
     /**
      * @param string $workspaceName
      * @param string $indexPostfix
+     * @throws \Exception
      */
-    protected function indexWorkspace($workspaceName, $indexPostfix)
+    protected function indexWorkspace(string $workspaceName, string $indexPostfix): void
     {
         $this->outputLine('<info>++</info> Indexing %s workspace', [$workspaceName]);
         $nodeCounter = 0;
@@ -277,30 +295,36 @@ protected function indexWorkspace($workspaceName, $indexPostfix)
      * @param string $indexPostfix
      * @return string
      * @throws \Flowpack\ElasticSearch\ContentRepositoryAdaptor\Exception
+     * @throws ConfigurationException
+     * @throws \Flowpack\ElasticSearch\Exception
+     * @throws \Neos\Flow\Http\Exception
      */
-    protected function createNextIndex($indexPostfix)
+    protected function createNextIndex(string $indexPostfix): string
     {
         $this->nodeIndexer->setIndexNamePostfix($indexPostfix);
         $this->nodeIndexer->getIndex()->create();
-        $this->log(sprintf('action=indexing step=index-created index=%s', $this->nodeIndexer->getIndexName()), LOG_INFO);
+        $this->logger->info(sprintf('Index %s created', $this->nodeIndexer->getIndexName()), LogEnvironment::fromMethodName(__METHOD__));
 
         return $this->nodeIndexer->getIndexName();
     }
 
     /**
      * Update Index Mapping
      *
+     * @param string $indexPostfix
      * @return void
+     * @throws ConfigurationException
      * @throws \Flowpack\ElasticSearch\ContentRepositoryAdaptor\Exception
+     * @throws \Flowpack\ElasticSearch\Exception
      */
-    protected function updateMapping($indexPostfix)
+    protected function updateMapping(string $indexPostfix): void
     {
         $nodeTypeMappingCollection = $this->nodeTypeMappingBuilder->buildMappingInformation($this->nodeIndexer->getIndex());
         foreach ($nodeTypeMappingCollection as $mapping) {
             $this->nodeIndexer->setIndexNamePostfix($indexPostfix);
             /** @var Mapping $mapping */
             $mapping->apply();
         }
-        $this->log(sprintf('action=indexing step=mapping-updated index=%s', $this->nodeIndexer->getIndexName()), LOG_INFO);
+        $this->logger->info(sprintf('Mapping updated for index %s', $this->nodeIndexer->getIndexName()), LogEnvironment::fromMethodName(__METHOD__));
     }
 }

Classes/Domain/Repository/NodeDataRepository.php

@@ -1,4 +1,6 @@
 <?php
+declare(strict_types=1);
+
 namespace Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\Domain\Repository;
 
 /*
@@ -12,6 +14,8 @@
  */
 
 use Doctrine\Common\Persistence\ObjectManager;
+use Doctrine\ORM\EntityManager;
+use Doctrine\ORM\EntityManagerInterface;
 use Doctrine\ORM\Internal\Hydration\IterableResult;
 use Doctrine\ORM\QueryBuilder;
 use Neos\ContentRepository\Domain\Model\NodeData;
@@ -23,11 +27,11 @@
  */
 class NodeDataRepository extends Repository
 {
-    const ENTITY_CLASSNAME = NodeData::class;
+    public const ENTITY_CLASSNAME = NodeData::class;
 
     /**
      * @Flow\Inject
-     * @var ObjectManager
+     * @var EntityManagerInterface
      */
     protected $entityManager;
 
@@ -37,7 +41,7 @@ class NodeDataRepository extends Repository
      * @param integer $maxResults
      * @return IterableResult
      */
-    public function findAllBySiteAndWorkspace($workspaceName, $firstResult = 0, $maxResults = 1000)
+    public function findAllBySiteAndWorkspace($workspaceName, $firstResult = 0, $maxResults = 1000): IterableResult
     {
         /** @var QueryBuilder $queryBuilder */
         $queryBuilder = $this->entityManager->createQueryBuilder();

Classes/Domain/Service/FakeNodeDataFactory.php

@@ -1,4 +1,6 @@
 <?php
+declare(strict_types=1);
+
 namespace Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\Domain\Service;
 
 /*

Classes/Indexer/NodeIndexer.php

@@ -1,4 +1,6 @@
 <?php
+declare(strict_types=1);
+
 namespace Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\Indexer;
 
 /*
@@ -21,7 +23,7 @@
 use Neos\Flow\Persistence\PersistenceManagerInterface;
 
 /**
- * Nodeindexer for use in batch jobs
+ * NodeIndexer for use in batch jobs
  */
 class NodeIndexer extends ContentRepositoryAdaptor\Indexer\NodeIndexer
 {
@@ -46,9 +48,9 @@ class NodeIndexer extends ContentRepositoryAdaptor\Indexer\NodeIndexer
     /**
      * @param NodeInterface $node
      * @param string|null $targetWorkspaceName In case indexing is triggered during publishing, a target workspace name will be passed in
-     * @throws \Neos\ContentRepository\Search\Exception\IndexingException
+     * @throws ContentRepositoryAdaptor\Exception
      */
-    public function indexNode(NodeInterface $node, $targetWorkspaceName = null)
+    public function indexNode(NodeInterface $node, $targetWorkspaceName = null): void
     {
         if( $node->isRemoved() ){
             $this->removeNode($node, $targetWorkspaceName);
@@ -77,8 +79,12 @@ public function indexNode(NodeInterface $node, $targetWorkspaceName = null)
     /**
      * @param NodeInterface $node
      * @param string|null $targetWorkspaceName In case indexing is triggered during publishing, a target workspace name will be passed in
+     * @throws ContentRepositoryAdaptor\Exception
+     * @throws \Flowpack\ElasticSearch\Exception
+     * @throws \Neos\Flow\Persistence\Exception\IllegalObjectTypeException
+     * @throws \Neos\Utility\Exception\FilesException
      */
-    public function removeNode(NodeInterface $node, $targetWorkspaceName = null)
+    public function removeNode(NodeInterface $node, string $targetWorkspaceName = null): void
     {
         if ($this->enableLiveAsyncIndexing !== true) {
             parent::removeNode($node, $targetWorkspaceName);
@@ -106,7 +112,7 @@ public function removeNode(NodeInterface $node, $targetWorkspaceName = null)
      * @param NodeInterface $node
      * @return array
      */
-    protected function nodeAsArray(NodeInterface $node)
+    protected function nodeAsArray(NodeInterface $node): array
     {
         return [
             [

Classes/IndexingJob.php

@@ -1,4 +1,6 @@
 <?php
+declare(strict_types=1);
+
 namespace Flowpack\ElasticSearch\ContentRepositoryQueueIndexer;
 
 /*
@@ -15,6 +17,7 @@
 use Flowpack\JobQueue\Common\Queue\QueueInterface;
 use Neos\ContentRepository\Domain\Model\NodeData;
 use Neos\ContentRepository\Domain\Model\NodeInterface;
+use Neos\Flow\Log\Utility\LogEnvironment;
 
 /**
  * Elasticsearch Node Indexing Job
@@ -34,13 +37,14 @@ public function execute(QueueInterface $queue, Message $message): bool
         $this->nodeIndexer->withBulkProcessing(function () {
             $numberOfNodes = count($this->nodes);
             $startTime = microtime(true);
+
             foreach ($this->nodes as $node) {
                 /** @var NodeData $nodeData */
                 $nodeData = $this->nodeDataRepository->findByIdentifier($node['persistenceObjectIdentifier']);
 
                 // Skip this iteration if the nodedata can not be fetched (deleted node)
                 if (!$nodeData instanceof NodeData) {
-                    $this->log(sprintf('action=indexing step=skipped node=%s message="Node data could not be loaded"', $node['identifier']), \LOG_NOTICE);
+                    $this->logger->notice(sprintf('Node data of node %s could not be loaded. Node might be deleted."', $node['identifier']), LogEnvironment::fromMethodName(__METHOD__));
                     continue;
                 }
 
@@ -54,7 +58,7 @@ public function execute(QueueInterface $queue, Message $message): bool
 
                 // Skip this iteration if the node can not be fetched from the current context
                 if (!$currentNode instanceof NodeInterface) {
-                    $this->log(sprintf('action=indexing step=failed node=%s message="Node could not be processed"', $node['identifier']), \LOG_WARNING);
+                    $this->logger->warning(sprintf('Node %s could not be processed"', $node['identifier']), LogEnvironment::fromMethodName(__METHOD__));
                     continue;
                 }
 
@@ -65,7 +69,7 @@ public function execute(QueueInterface $queue, Message $message): bool
             $this->nodeIndexer->flush();
             $duration = microtime(true) - $startTime;
             $rate = $numberOfNodes / $duration;
-            $this->log(sprintf('action=indexing step=finished number_of_nodes=%d duration=%f nodes_per_second=%f', $numberOfNodes, $duration, $rate), \LOG_INFO);
+            $this->logger->info(sprintf('Indexed %s nodes in %s seconds (%s nodes per second)', $numberOfNodes, $duration, $rate), LogEnvironment::fromMethodName(__METHOD__));
         });
 
         return true;