Skip to content

Commit 113138a

Browse files
dfeyerdfeyer
authored
Merge pull request #15 from kdambekalns/bugfix/14-realtime-removal
BUGFIX: Fix realtime indexing of node deletion
2 parents db61678 + 9818f9a commit 113138a

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,12 +7,14 @@
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;
14+
use Neos\ContentRepository\Domain\Repository\WorkspaceRepository;
1215
use Neos\Flow\Annotations as Flow;
1316
use Neos\Flow\Cli\CommandController;
1417
use Neos\Flow\Persistence\PersistenceManagerInterface;
15-
use Neos\ContentRepository\Domain\Repository\WorkspaceRepository;
1618
use Neos\Utility\Files;
1719

1820
/**
@@ -26,7 +28,6 @@ class NodeIndexQueueCommandController extends CommandController
2628

2729
const BATCH_QUEUE_NAME = 'Flowpack.ElasticSearch.ContentRepositoryQueueIndexer';
2830
const LIVE_QUEUE_NAME = 'Flowpack.ElasticSearch.ContentRepositoryQueueIndexer.Live';
29-
const DEFAULT_BATCH_SIZE = 500;
3031

3132
/**
3233
* @var JobManager
@@ -71,23 +72,25 @@ class NodeIndexQueueCommandController extends CommandController
7172
protected $nodeIndexer;
7273

7374
/**
74-
* @Flow\InjectConfiguration(package="Flowpack.ElasticSearch.ContentRepositoryQueueIndexer")
75-
* @var array
75+
* @Flow\InjectConfiguration(path="batchSize")
76+
* @var int
7677
*/
77-
protected $settings;
78+
protected $batchSize;
7879

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

90-
9194
$this->outputLine();
9295
$this->outputLine('<b>Indexing on %s ...</b>', [$indexName]);
9396

@@ -121,6 +124,7 @@ public function buildCommand($workspace = null)
121124
* @param int $limit If set, only the given amount of jobs are processed (successful or not) before the script exits
122125
* @param bool $verbose Output debugging information
123126
* @return void
127+
* @throws \Neos\Flow\Mvc\Exception\StopActionException
124128
*/
125129
public function workCommand($queue = 'batch', $exitAfter = null, $limit = null, $verbose = false)
126130
{
@@ -152,8 +156,8 @@ public function workCommand($queue = 'batch', $exitAfter = null, $limit = null,
152156
}
153157
try {
154158
$message = $this->jobManager->waitAndExecute($queueName, $timeout);
155-
} catch (JobQueueException $exception) {
156-
$numberOfJobExecutions ++;
159+
} catch (Exception $exception) {
160+
$numberOfJobExecutions++;
157161
$this->outputLine('<error>%s</error>', [$exception->getMessage()]);
158162
if ($verbose && $exception->getPrevious() instanceof \Exception) {
159163
$this->outputLine(' Reason: %s', [$exception->getPrevious()->getMessage()]);
@@ -163,7 +167,7 @@ public function workCommand($queue = 'batch', $exitAfter = null, $limit = null,
163167
$this->quit(1);
164168
}
165169
if ($message !== null) {
166-
$numberOfJobExecutions ++;
170+
$numberOfJobExecutions++;
167171
if ($verbose) {
168172
$messagePayload = strlen($message->getPayload()) <= 50 ? $message->getPayload() : substr($message->getPayload(), 0, 50) . '...';
169173
$this->outputLine('<success>Successfully executed job "%s" (%s)</success>', [$message->getIdentifier(), $messagePayload]);
@@ -181,7 +185,6 @@ public function workCommand($queue = 'batch', $exitAfter = null, $limit = null,
181185
}
182186
$this->quit();
183187
}
184-
185188
} while (true);
186189
}
187190

@@ -190,8 +193,12 @@ public function workCommand($queue = 'batch', $exitAfter = null, $limit = null,
190193
*/
191194
public function flushCommand()
192195
{
193-
$this->queueManager->getQueue(self::BATCH_QUEUE_NAME)->flush();
194-
$this->outputSystemReport();
196+
try {
197+
$this->queueManager->getQueue(self::BATCH_QUEUE_NAME)->flush();
198+
$this->outputSystemReport();
199+
} catch (Exception $exception) {
200+
$this->outputLine('An error occurred: %s', [$exception->getMessage()]);
201+
}
195202
$this->outputLine();
196203
}
197204

@@ -205,7 +212,11 @@ protected function outputSystemReport()
205212
$time = microtime(true) - $_SERVER["REQUEST_TIME_FLOAT"];
206213
$this->outputLine('Execution time : %s seconds', [$time]);
207214
$this->outputLine('Indexing Queue : %s', [self::BATCH_QUEUE_NAME]);
208-
$this->outputLine('Pending Jobs : %s', [$this->queueManager->getQueue(self::BATCH_QUEUE_NAME)->count()]);
215+
try {
216+
$this->outputLine('Pending Jobs : %s', [$this->queueManager->getQueue(self::BATCH_QUEUE_NAME)->count()]);
217+
} catch (Exception $exception) {
218+
$this->outputLine('Pending Jobs : Error, queue not found, %s', [$exception->getMessage()]);
219+
}
209220
}
210221

211222
/**
@@ -217,16 +228,19 @@ protected function indexWorkspace($workspaceName, $indexPostfix)
217228
$this->outputLine('<info>++</info> Indexing %s workspace', [$workspaceName]);
218229
$nodeCounter = 0;
219230
$offset = 0;
220-
$batchSize = $this->settings['batchSize'] ?? static::DEFAULT_BATCH_SIZE;
221231
while (true) {
222-
$iterator = $this->nodeDataRepository->findAllBySiteAndWorkspace($workspaceName, $offset, $batchSize);
232+
$iterator = $this->nodeDataRepository->findAllBySiteAndWorkspace($workspaceName, $offset, $this->batchSize);
223233

224234
$jobData = [];
225235

226236
foreach ($this->nodeDataRepository->iterate($iterator) as $data) {
227237
$jobData[] = [
228-
'nodeIdentifier' => $data['nodeIdentifier'],
229-
'dimensions' => $data['dimensions']
238+
'persistenceObjectIdentifier' => $data['persistenceObjectIdentifier'],
239+
'identifier' => $data['identifier'],
240+
'dimensions' => $data['dimensions'],
241+
'workspace' => $workspaceName,
242+
'nodeType' => $data['nodeType'],
243+
'path' => $data['path'],
230244
];
231245
$nodeCounter++;
232246
}
@@ -238,7 +252,7 @@ protected function indexWorkspace($workspaceName, $indexPostfix)
238252
$indexingJob = new IndexingJob($indexPostfix, $workspaceName, $jobData);
239253
$this->jobManager->queue(self::BATCH_QUEUE_NAME, $indexingJob);
240254
$this->output('.');
241-
$offset += $batchSize;
255+
$offset += $this->batchSize;
242256
$this->persistenceManager->clearState();
243257
}
244258
$this->outputLine();
@@ -249,17 +263,22 @@ protected function indexWorkspace($workspaceName, $indexPostfix)
249263
/**
250264
* @param string $indexPostfix
251265
* @return string
266+
* @throws \Flowpack\ElasticSearch\ContentRepositoryAdaptor\Exception
252267
*/
253268
protected function createNextIndex($indexPostfix)
254269
{
255270
$this->nodeIndexer->setIndexNamePostfix($indexPostfix);
256271
$this->nodeIndexer->getIndex()->create();
257272
$this->log(sprintf('action=indexing step=index-created index=%s', $this->nodeIndexer->getIndexName()), LOG_INFO);
273+
258274
return $this->nodeIndexer->getIndexName();
259275
}
260276

261277
/**
262278
* Update Index Mapping
279+
*
280+
* @return void
281+
* @throws \Flowpack\ElasticSearch\ContentRepositoryAdaptor\Exception
263282
*/
264283
protected function updateMapping()
265284
{

‎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)