Skip to content

Commit 4c065f5

Browse files
dfeyerdfeyer
committed
Merge branch '3.0'
# Conflicts: # Classes/Command/NodeIndexQueueCommandController.php
2 parents f9fce86 + 113138a commit 4c065f5

File tree

9 files changed

+346
-116
lines changed

9 files changed

+346
-116
lines changed

‎Classes/AbstractIndexingJob.php‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
<?php
2+
namespace Flowpack\ElasticSearch\ContentRepositoryQueueIndexer;
3+
4+
use Flowpack\ElasticSearch\ContentRepositoryAdaptor\Indexer\NodeIndexer;
5+
use Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\Domain\Repository\NodeDataRepository;
6+
use Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\Domain\Service\FakeNodeDataFactory;
7+
use Flowpack\JobQueue\Common\Job\JobInterface;
8+
use Neos\ContentRepository\Domain\Factory\NodeFactory;
9+
use Neos\ContentRepository\Domain\Service\ContextFactoryInterface;
10+
use Neos\Flow\Annotations as Flow;
11+
use Neos\Flow\Utility\Algorithms;
12+
13+
/**
14+
* Elasticsearch Node Abstract Job
15+
*/
16+
abstract class AbstractIndexingJob implements JobInterface
17+
{
18+
use LoggerTrait;
19+
20+
/**
21+
* @var NodeIndexer
22+
* @Flow\Inject
23+
*/
24+
protected $nodeIndexer;
25+
26+
/**
27+
* @var NodeDataRepository
28+
* @Flow\Inject
29+
*/
30+
protected $nodeDataRepository;
31+
32+
/**
33+
* @var NodeFactory
34+
* @Flow\Inject
35+
*/
36+
protected $nodeFactory;
37+
38+
/**
39+
* @var ContextFactoryInterface
40+
* @Flow\Inject
41+
*/
42+
protected $contextFactory;
43+
44+
/**
45+
* @var FakeNodeDataFactory
46+
* @Flow\Inject
47+
*/
48+
protected $fakeNodeDataFactory;
49+
50+
/**
51+
* @var string
52+
*/
53+
protected $identifier;
54+
55+
/**
56+
* @var string
57+
*/
58+
protected $targetWorkspaceName;
59+
60+
/**
61+
* @var string
62+
*/
63+
protected $indexPostfix;
64+
65+
/**
66+
* @var array
67+
*/
68+
protected $nodes = [];
69+
70+
/**
71+
* @param string $indexPostfix
72+
* @param string $targetWorkspaceName In case indexing is triggered during publishing, a target workspace name will be passed in
73+
* @param array $nodes
74+
*/
75+
public function __construct($indexPostfix, $targetWorkspaceName, array $nodes)
76+
{
77+
$this->identifier = Algorithms::generateRandomString(24);
78+
$this->targetWorkspaceName = $targetWorkspaceName;
79+
$this->indexPostfix = $indexPostfix;
80+
$this->nodes = $nodes;
81+
}
82+
83+
/**
84+
* Get an optional identifier for the job
85+
*
86+
* @return string A job identifier
87+
*/
88+
public function getIdentifier()
89+
{
90+
return $this->identifier;
91+
}
92+
}

‎Classes/Command/NodeIndexQueueCommandController.php‎

Lines changed: 37 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -7,14 +7,16 @@
77
use Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\IndexingJob;
88
use Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\LoggerTrait;
99
use Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\UpdateAliasJob;
10+
use Flowpack\ElasticSearch\Domain\Model\Mapping;
11+
use Flowpack\JobQueue\Common\Exception;
1012
use Flowpack\JobQueue\Common\Job\JobManager;
1113
use Flowpack\JobQueue\Common\Queue\QueueManager;
1214
use Flowpack\ElasticSearch\Domain\Model\Mapping;
1315
use Flowpack\JobQueue\Common\Exception as JobQueueException;
16+
use Neos\ContentRepository\Domain\Repository\WorkspaceRepository;
1417
use Neos\Flow\Annotations as Flow;
1518
use Neos\Flow\Cli\CommandController;
1619
use Neos\Flow\Persistence\PersistenceManagerInterface;
17-
use Neos\ContentRepository\Domain\Repository\WorkspaceRepository;
1820
use Neos\Utility\Files;
1921

2022
/**
@@ -28,7 +30,6 @@ class NodeIndexQueueCommandController extends CommandController
2830

2931
const BATCH_QUEUE_NAME = 'Flowpack.ElasticSearch.ContentRepositoryQueueIndexer';
3032
const LIVE_QUEUE_NAME = 'Flowpack.ElasticSearch.ContentRepositoryQueueIndexer.Live';
31-
const DEFAULT_BATCH_SIZE = 500;
3233

3334
/**
3435
* @var JobManager
@@ -73,23 +74,25 @@ class NodeIndexQueueCommandController extends CommandController
7374
protected $nodeIndexer;
7475

7576
/**
76-
* @Flow\InjectConfiguration(package="Flowpack.ElasticSearch.ContentRepositoryQueueIndexer")
77-
* @var array
77+
* @Flow\InjectConfiguration(path="batchSize")
78+
* @var int
7879
*/
79-
protected $settings;
80+
protected $batchSize;
8081

8182
/**
8283
* Index all nodes by creating a new index and when everything was completed, switch the index alias.
8384
*
8485
* @param string $workspace
86+
* @throws \Flowpack\JobQueue\Common\Exception
87+
* @throws \Neos\Flow\Mvc\Exception\StopActionException
88+
* @throws \Flowpack\ElasticSearch\ContentRepositoryAdaptor\Exception
8589
*/
8690
public function buildCommand($workspace = null)
8791
{
8892
$indexPostfix = time();
8993
$indexName = $this->createNextIndex($indexPostfix);
9094
$this->updateMapping();
9195

92-
9396
$this->outputLine();
9497
$this->outputLine('<b>Indexing on %s ...</b>', [$indexName]);
9598

@@ -123,6 +126,7 @@ public function buildCommand($workspace = null)
123126
* @param int $limit If set, only the given amount of jobs are processed (successful or not) before the script exits
124127
* @param bool $verbose Output debugging information
125128
* @return void
129+
* @throws \Neos\Flow\Mvc\Exception\StopActionException
126130
*/
127131
public function workCommand($queue = 'batch', $exitAfter = null, $limit = null, $verbose = false)
128132
{
@@ -154,8 +158,8 @@ public function workCommand($queue = 'batch', $exitAfter = null, $limit = null,
154158
}
155159
try {
156160
$message = $this->jobManager->waitAndExecute($queueName, $timeout);
157-
} catch (JobQueueException $exception) {
158-
$numberOfJobExecutions ++;
161+
} catch (Exception $exception) {
162+
$numberOfJobExecutions++;
159163
$this->outputLine('<error>%s</error>', [$exception->getMessage()]);
160164
if ($verbose && $exception->getPrevious() instanceof \Exception) {
161165
$this->outputLine(' Reason: %s', [$exception->getPrevious()->getMessage()]);
@@ -165,7 +169,7 @@ public function workCommand($queue = 'batch', $exitAfter = null, $limit = null,
165169
$this->quit(1);
166170
}
167171
if ($message !== null) {
168-
$numberOfJobExecutions ++;
172+
$numberOfJobExecutions++;
169173
if ($verbose) {
170174
$messagePayload = strlen($message->getPayload()) <= 50 ? $message->getPayload() : substr($message->getPayload(), 0, 50) . '...';
171175
$this->outputLine('<success>Successfully executed job "%s" (%s)</success>', [$message->getIdentifier(), $messagePayload]);
@@ -183,7 +187,6 @@ public function workCommand($queue = 'batch', $exitAfter = null, $limit = null,
183187
}
184188
$this->quit();
185189
}
186-
187190
} while (true);
188191
}
189192

@@ -192,8 +195,12 @@ public function workCommand($queue = 'batch', $exitAfter = null, $limit = null,
192195
*/
193196
public function flushCommand()
194197
{
195-
$this->queueManager->getQueue(self::BATCH_QUEUE_NAME)->flush();
196-
$this->outputSystemReport();
198+
try {
199+
$this->queueManager->getQueue(self::BATCH_QUEUE_NAME)->flush();
200+
$this->outputSystemReport();
201+
} catch (Exception $exception) {
202+
$this->outputLine('An error occurred: %s', [$exception->getMessage()]);
203+
}
197204
$this->outputLine();
198205
}
199206

@@ -207,7 +214,11 @@ protected function outputSystemReport()
207214
$time = microtime(true) - $_SERVER["REQUEST_TIME_FLOAT"];
208215
$this->outputLine('Execution time : %s seconds', [$time]);
209216
$this->outputLine('Indexing Queue : %s', [self::BATCH_QUEUE_NAME]);
210-
$this->outputLine('Pending Jobs : %s', [$this->queueManager->getQueue(self::BATCH_QUEUE_NAME)->count()]);
217+
try {
218+
$this->outputLine('Pending Jobs : %s', [$this->queueManager->getQueue(self::BATCH_QUEUE_NAME)->count()]);
219+
} catch (Exception $exception) {
220+
$this->outputLine('Pending Jobs : Error, queue not found, %s', [$exception->getMessage()]);
221+
}
211222
}
212223

213224
/**
@@ -219,16 +230,19 @@ protected function indexWorkspace($workspaceName, $indexPostfix)
219230
$this->outputLine('<info>++</info> Indexing %s workspace', [$workspaceName]);
220231
$nodeCounter = 0;
221232
$offset = 0;
222-
$batchSize = $this->settings['batchSize'] ?? static::DEFAULT_BATCH_SIZE;
223233
while (true) {
224-
$iterator = $this->nodeDataRepository->findAllBySiteAndWorkspace($workspaceName, $offset, $batchSize);
234+
$iterator = $this->nodeDataRepository->findAllBySiteAndWorkspace($workspaceName, $offset, $this->batchSize);
225235

226236
$jobData = [];
227237

228238
foreach ($this->nodeDataRepository->iterate($iterator) as $data) {
229239
$jobData[] = [
230-
'nodeIdentifier' => $data['nodeIdentifier'],
231-
'dimensions' => $data['dimensions']
240+
'persistenceObjectIdentifier' => $data['persistenceObjectIdentifier'],
241+
'identifier' => $data['identifier'],
242+
'dimensions' => $data['dimensions'],
243+
'workspace' => $workspaceName,
244+
'nodeType' => $data['nodeType'],
245+
'path' => $data['path'],
232246
];
233247
$nodeCounter++;
234248
}
@@ -240,7 +254,7 @@ protected function indexWorkspace($workspaceName, $indexPostfix)
240254
$indexingJob = new IndexingJob($indexPostfix, $workspaceName, $jobData);
241255
$this->jobManager->queue(self::BATCH_QUEUE_NAME, $indexingJob);
242256
$this->output('.');
243-
$offset += $batchSize;
257+
$offset += $this->batchSize;
244258
$this->persistenceManager->clearState();
245259
}
246260
$this->outputLine();
@@ -251,17 +265,22 @@ protected function indexWorkspace($workspaceName, $indexPostfix)
251265
/**
252266
* @param string $indexPostfix
253267
* @return string
268+
* @throws \Flowpack\ElasticSearch\ContentRepositoryAdaptor\Exception
254269
*/
255270
protected function createNextIndex($indexPostfix)
256271
{
257272
$this->nodeIndexer->setIndexNamePostfix($indexPostfix);
258273
$this->nodeIndexer->getIndex()->create();
259274
$this->log(sprintf('action=indexing step=index-created index=%s', $this->nodeIndexer->getIndexName()), LOG_INFO);
275+
260276
return $this->nodeIndexer->getIndexName();
261277
}
262278

263279
/**
264280
* Update Index Mapping
281+
*
282+
* @return void
283+
* @throws \Flowpack\ElasticSearch\ContentRepositoryAdaptor\Exception
265284
*/
266285
protected function updateMapping()
267286
{

‎Classes/Domain/Repository/NodeDataRepository.php‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
use Doctrine\Common\Persistence\ObjectManager;
55
use Doctrine\ORM\Internal\Hydration\IterableResult;
66
use Doctrine\ORM\QueryBuilder;
7+
use Neos\ContentRepository\Domain\Model\NodeData;
78
use Neos\Flow\Annotations as Flow;
89
use Neos\Flow\Persistence\Repository;
910

@@ -12,7 +13,7 @@
1213
*/
1314
class NodeDataRepository extends Repository
1415
{
15-
const ENTITY_CLASSNAME = 'Neos\ContentRepository\Domain\Model\NodeData';
16+
const ENTITY_CLASSNAME = NodeData::class;
1617

1718
/**
1819
* @Flow\Inject
@@ -28,12 +29,11 @@ class NodeDataRepository extends Repository
2829
*/
2930
public function findAllBySiteAndWorkspace($workspaceName, $firstResult = 0, $maxResults = 1000)
3031
{
31-
3232
/** @var QueryBuilder $queryBuilder */
3333
$queryBuilder = $this->entityManager->createQueryBuilder();
3434

35-
$queryBuilder->select('n.Persistence_Object_Identifier nodeIdentifier, n.dimensionValues dimensions')
36-
->from('Neos\ContentRepository\Domain\Model\NodeData', 'n')
35+
$queryBuilder->select('n.Persistence_Object_Identifier persistenceObjectIdentifier, n.identifier identifier, n.dimensionValues dimensions, n.nodeType nodeType, n.path path')
36+
->from(NodeData::class, 'n')
3737
->where("n.workspace = :workspace AND n.removed = :removed AND n.movedTo IS NULL")
3838
->setFirstResult((integer)$firstResult)
3939
->setMaxResults((integer)$maxResults)
@@ -64,7 +64,7 @@ public function iterate(IterableResult $iterator, callable $callback = null)
6464
if ($callback !== null) {
6565
call_user_func($callback, $iteration, $object);
6666
}
67-
++$iteration;
67+
$iteration++;
6868
}
6969
}
7070
}

‎Classes/Domain/Service/FakeNodeDataFactory.php‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,69 @@
1+
<?php
2+
namespace Flowpack\ElasticSearch\ContentRepositoryQueueIndexer\Domain\Service;
3+
4+
use Flowpack\ElasticSearch\ContentRepositoryAdaptor\Exception;
5+
use Neos\ContentRepository\Domain\Model\NodeData;
6+
use Neos\ContentRepository\Domain\Repository\WorkspaceRepository;
7+
use Neos\ContentRepository\Domain\Service\NodeTypeManager;
8+
use Neos\ContentRepository\Exception\NodeTypeNotFoundException;
9+
use Neos\Flow\Annotations as Flow;
10+
11+
/**
12+
* @Flow\Scope("singleton")
13+
*/
14+
class FakeNodeDataFactory
15+
{
16+
/**
17+
* @var WorkspaceRepository
18+
* @Flow\Inject
19+
*/
20+
protected $workspaceRepository;
21+
22+
/**
23+
* @var NodeTypeManager
24+
* @Flow\Inject
25+
*/
26+
protected $nodeTypeManager;
27+
28+
/**
29+
* This creates a "fake" removed NodeData instance from the given payload
30+
*
31+
* @param array $payload
32+
* @return NodeData
33+
* @throws Exception
34+
*/
35+
public function createFromPayload(array $payload)
36+
{
37+
if (!isset($payload['workspace']) || empty($payload['workspace'])) {
38+
throw new Exception('Unable to create fake node data, missing workspace value', 1508448007);
39+
}
40+
if (!isset($payload['path']) || empty($payload['path'])) {
41+
throw new Exception('Unable to create fake node data, missing path value', 1508448008);
42+
}
43+
if (!isset($payload['identifier']) || empty($payload['identifier'])) {
44+
throw new Exception('Unable to create fake node data, missing identifier value', 1508448009);
45+
}
46+
if (!isset($payload['nodeType']) || empty($payload['nodeType'])) {
47+
throw new Exception('Unable to create fake node data, missing nodeType value', 1508448011);
48+
}
49+
50+
$workspace = $this->workspaceRepository->findOneByName($payload['workspace']);
51+
if ($workspace === null) {
52+
throw new Exception('Unable to create fake node data, workspace not found', 1508448028);
53+
}
54+
55+
$nodeData = new NodeData($payload['path'], $workspace, $payload['identifier'], isset($payload['dimensions']) ? $payload['dimensions'] : null);
56+
try {
57+
$nodeData->setNodeType($this->nodeTypeManager->getNodeType($payload['nodeType']));
58+
} catch (NodeTypeNotFoundException $e) {
59+
throw new Exception('Unable to create fake node data, node type not found', 1509362172);
60+
}
61+
62+
$nodeData->setProperty('title', 'Fake node');
63+
$nodeData->setProperty('uriPathSegment', 'fake-node');
64+
65+
$nodeData->setRemoved(true);
66+
67+
return $nodeData;
68+
}
69+
}

0 commit comments

Comments
 (0)