Skip to content

Commit 2995b54

Browse files
makasimmakasim
committed
add configuration.
1 parent 8119cc4 commit 2995b54

File tree

4 files changed

+82
-19
lines changed

4 files changed

+82
-19
lines changed

‎DependencyInjection/Configuration.php

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
<?php
2+
3+
namespace Enqueue\Bundle\DependencyInjection;
4+
5+
use Symfony\Component\Config\Definition\Builder\TreeBuilder;
6+
use Symfony\Component\Config\Definition\ConfigurationInterface;
7+
8+
class Configuration implements ConfigurationInterface
9+
{
10+
/**
11+
* {@inheritdoc}
12+
*/
13+
public function getConfigTreeBuilder()
14+
{
15+
$tb = new TreeBuilder();
16+
$rootNode = $tb->root('enqueue_elastica');
17+
$rootNode
18+
->children()
19+
->arrayNode('doctrine_queue_listeners')
20+
->prototype('array')
21+
->addDefaultsIfNotSet()
22+
->children()
23+
->booleanNode('on_insert')->defaultTrue()->end()
24+
->booleanNode('on_update')->defaultTrue()->end()
25+
->booleanNode('on_remove')->defaultTrue()->end()
26+
->scalarNode('index_name')->isRequired()->cannotBeEmpty()->end()
27+
->scalarNode('type_name')->isRequired()->cannotBeEmpty()->end()
28+
->scalarNode('model_class')->isRequired()->cannotBeEmpty()->end()
29+
;
30+
}
31+
}

‎DependencyInjection/EnqueueElasticaExtension.php

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,11 @@
22

33
namespace Enqueue\ElasticaBundle\DependencyInjection;
44

5+
use Enqueue\ElasticaBundle\Doctrine\SyncIndexWithObjectChangeListener;
56
use Symfony\Component\Config\FileLocator;
67
use Symfony\Component\DependencyInjection\ContainerBuilder;
78
use Symfony\Component\DependencyInjection\Loader\YamlFileLoader;
9+
use Symfony\Component\DependencyInjection\Reference;
810
use Symfony\Component\HttpKernel\DependencyInjection\Extension;
911

1012
class EnqueueElasticaExtension extends Extension
@@ -14,7 +16,27 @@ class EnqueueElasticaExtension extends Extension
1416
*/
1517
public function load(array $configs, ContainerBuilder $container)
1618
{
19+
$configuration = $this->getConfiguration($configs, $container);
20+
$config = $this->processConfiguration($configuration, $configs);
21+
1722
$loader = new YamlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
1823
$loader->load('services.yml');
24+
25+
if (isset($config['doctrine_queue_listeners'])) {
26+
foreach ($config['doctrine_queue_listeners'] as $listenerConfig) {
27+
$listenerId = sprintf(
28+
'enqueue_elastica.doctrine_queue_listener.%s.%s',
29+
$listenerConfig['index_name'],
30+
$listenerConfig['type_name']
31+
);
32+
33+
$container->register($listenerId, SyncIndexWithObjectChangeListener::class)
34+
->addArgument(new Reference('enqueue.transport.context'))
35+
->addArgument($listenerConfig['modelClass'])
36+
->addArgument($listenerConfig)
37+
->addTag('doctrine.event_subscriber')
38+
;
39+
}
40+
}
1941
}
2042
}

‎Queue/SyncIndexWithDoctrineORMObjectChangeProcessor.php

Lines changed: 20 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
use Enqueue\Client\CommandSubscriberInterface;
55
use Enqueue\Consumption\QueueSubscriberInterface;
66
use Enqueue\Util\JSON;
7-
use FOS\ElasticaBundle\Persister\ObjectPersisterInterface;
7+
use FOS\ElasticaBundle\Persister\PersisterRegistry;
88
use FOS\ElasticaBundle\Provider\IndexableInterface;
99
use Interop\Queue\PsrContext;
1010
use Interop\Queue\PsrMessage;
@@ -14,9 +14,9 @@
1414
final class SyncIndexWithDoctrineORMObjectChangeProcessor implements PsrProcessor, CommandSubscriberInterface, QueueSubscriberInterface
1515
{
1616
/**
17-
* @var ObjectPersisterInterface
17+
* @var PersisterRegistry
1818
*/
19-
private $objectPersister;
19+
private $persisterRegistry;
2020

2121
/**
2222
* @var IndexableInterface
@@ -28,9 +28,9 @@ final class SyncIndexWithDoctrineORMObjectChangeProcessor implements PsrProcesso
2828
*/
2929
private $doctrine;
3030

31-
public function __construct(RegistryInterface $doctrine, ObjectPersisterInterface $objectPersister, IndexableInterface $indexable)
31+
public function __construct(RegistryInterface $doctrine, PersisterRegistry $persisterRegistry, IndexableInterface $indexable)
3232
{
33-
$this->objectPersister = $objectPersister;
33+
$this->persisterRegistry = $persisterRegistry;
3434
$this->indexable = $indexable;
3535
$this->doctrine = $doctrine;
3636
}
@@ -46,42 +46,43 @@ public function process(PsrMessage $message, PsrContext $context)
4646
return self::REJECT;
4747
}
4848

49-
$indexName = $data['indexName'];
50-
$typeName = $data['typeName'];
49+
$index = $data['indexName'];
50+
$type = $data['typeName'];
5151

52-
$objectRepository = $this->doctrine->getManagerForClass($data['modelClass'])->getRepository($data['modelClass']);
52+
$repository = $this->doctrine->getManagerForClass($data['modelClass'])->getRepository($data['modelClass']);
53+
$persister = $this->persisterRegistry->getPersister($index, $type);
5354

5455
switch ($data['action']) {
5556
case 'update':
56-
if (false == $object = $objectRepository->find($data['id'])) {
57-
$this->objectPersister->deleteById($data['id']);
57+
if (false == $object = $repository->find($data['id'])) {
58+
$persister->deleteById($data['id']);
5859

5960
return self::REJECT;
6061
}
6162

62-
if ($this->objectPersister->handlesObject($object)) {
63-
if ($this->indexable->isObjectIndexable($indexName, $typeName, $object)) {
64-
$this->objectPersister->replaceOne($object);
63+
if ($persister->handlesObject($object)) {
64+
if ($this->indexable->isObjectIndexable($index, $type, $object)) {
65+
$persister->replaceOne($object);
6566
} else {
66-
$this->objectPersister->deleteOne($object);
67+
$persister->deleteOne($object);
6768
}
6869
}
6970

7071
break;
7172
case 'insert':
72-
if (false == $object = $objectRepository->find($data['id'])) {
73-
$this->objectPersister->deleteById($data['id']);
73+
if (false == $object = $repository->find($data['id'])) {
74+
$persister->deleteById($data['id']);
7475

7576
return self::REJECT;
7677
}
7778

78-
if ($this->objectPersister->handlesObject($object) && $this->indexable->isObjectIndexable($indexName, $typeName, $object)) {
79-
$this->objectPersister->insertOne($object);
79+
if ($persister->handlesObject($object) && $this->indexable->isObjectIndexable($index, $type, $object)) {
80+
$persister->insertOne($object);
8081
}
8182

8283
break;
8384
case 'delete':
84-
$this->objectPersister->deleteById($data['id']);
85+
$persister->deleteById($data['id']);
8586

8687
break;
8788
default:

‎Resources/config/services.yml

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,15 @@ services:
77
tags:
88
- { name: "enqueue.client.processor" }
99

10+
enqueue_elastica.sync_index_with_doctrine_orm_object_change_processor:
11+
class: 'Enqueue\ElasticaBundle\Queue\SyncIndexWithDoctrineORMObjectChangeProcessor'
12+
arguments:
13+
- '@doctrine'
14+
- '@fos_elastica.persister_registry'
15+
- '@fos_elastica.indexable'
16+
tags:
17+
- { name: "enqueue.client.processor" }
18+
1019
enqueue_elastica.purge_populate_queue_listener:
1120
class: 'Enqueue\ElasticaBundle\Persister\Listener\PurgePopulateQueueListener'
1221
arguments:

0 commit comments

Comments
 (0)