Skip to content

Commit 838c3a4

Browse files
makasimmakasim
committed
Upgrade to Enqueue 0.9
1 parent c195787 commit 838c3a4

9 files changed

+78
-156
lines changed

‎DependencyInjection/Configuration.php

+1-1
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ public function getConfigTreeBuilder()
1717
$rootNode
1818
->children()
1919
->booleanNode('enabled')->defaultValue(true)->end()
20-
->scalarNode('context')->defaultValue('enqueue.transport.context')->cannotBeEmpty()->end()
20+
->scalarNode('transport')->defaultValue('%enqueue.default_transport%')->cannotBeEmpty()->isRequired()->end()
2121
->arrayNode('doctrine')
2222
->children()
2323
->scalarNode('driver')->defaultValue('orm')->cannotBeEmpty()

‎DependencyInjection/EnqueueElasticaExtension.php

+41-22
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,14 @@
22

33
namespace Enqueue\ElasticaBundle\DependencyInjection;
44

5+
use Enqueue\ElasticaBundle\Doctrine\Queue\SyncIndexWithObjectChangeProcessor;
56
use Enqueue\ElasticaBundle\Doctrine\SyncIndexWithObjectChangeListener;
6-
use Symfony\Component\Config\FileLocator;
7+
use Enqueue\ElasticaBundle\Persister\Listener\PurgePopulateQueueListener;
8+
use Enqueue\ElasticaBundle\Persister\QueuePagerPersister;
9+
use Enqueue\ElasticaBundle\Queue\PopulateProcessor;
10+
use Enqueue\Symfony\DependencyInjection\TransportFactory;
11+
use Enqueue\Symfony\DiUtils;
712
use Symfony\Component\DependencyInjection\ContainerBuilder;
8-
use Symfony\Component\DependencyInjection\Loader\YamlFileLoader;
913
use Symfony\Component\DependencyInjection\Reference;
1014
use Symfony\Component\HttpKernel\DependencyInjection\Extension;
1115

@@ -22,13 +26,44 @@ public function load(array $configs, ContainerBuilder $container)
2226
return;
2327
}
2428

25-
$loader = new YamlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
26-
$loader->load('services.yml');
29+
$transport = $container->getParameterBag()->resolveValue($config['transport']);
2730

28-
$container->setAlias('enqueue_elastica.context', $config['context']);
31+
$diUtils = new DiUtils(TransportFactory::MODULE, $transport);
32+
$container->setAlias('enqueue_elastica.context', $diUtils->format('context'));
33+
34+
$container->register('enqueue_elastica.populate_processor', PopulateProcessor::class)
35+
->addArgument(new Reference('fos_elastica.pager_provider_registry'))
36+
->addArgument(new Reference('fos_elastica.pager_persister_registry'))
37+
38+
->addTag('enqueue.command_subscriber', ['client' => $transport])
39+
->addTag('enqueue.transport.processor', ['transport' => $transport])
40+
;
41+
42+
$container->register('enqueue_elastica.purge_populate_queue_listener', PurgePopulateQueueListener::class)
43+
->addArgument(new Reference('enqueue_elastica.context'))
44+
45+
->addTag('kernel.event_subscriber')
46+
;
47+
48+
$container->register('enqueue_elastica.queue_pager_perister', QueuePagerPersister::class)
49+
->addArgument(new Reference('enqueue_elastica.context'))
50+
->addArgument(new Reference('fos_elastica.persister_registry'))
51+
->addArgument(new Reference('event_dispatcher'))
52+
53+
->addTag('fos_elastica.pager_persister', ['persisterName' => 'queue'])
54+
;
2955

30-
$doctrineDriver = $config['doctrine']['driver'];
3156
if (false == empty($config['doctrine']['queue_listeners'])) {
57+
$doctrineDriver = $config['doctrine']['driver'];
58+
59+
$container->register('enqueue_elastica.doctrine.sync_index_with_object_change_processor', SyncIndexWithObjectChangeProcessor::class)
60+
->addArgument(new Reference($this->getManagerRegistry($doctrineDriver)))
61+
->addArgument(new Reference('fos_elastica.persister_registry'))
62+
->addArgument(new Reference('fos_elastica.indexable'))
63+
->addTag('enqueue.command_subscriber', ['client' => $transport])
64+
->addTag('enqueue.transport.processor', ['transport' => $transport])
65+
;
66+
3267
foreach ($config['doctrine']['queue_listeners'] as $listenerConfig) {
3368
$listenerId = sprintf(
3469
'enqueue_elastica.doctrine_queue_listener.%s.%s',
@@ -45,19 +80,8 @@ public function load(array $configs, ContainerBuilder $container)
4580
;
4681
}
4782
}
48-
49-
$serviceId = 'enqueue_elastica.doctrine.sync_index_with_object_change_processor';
50-
$managerRegistry = $this->getManagerRegistry($doctrineDriver);
51-
$container
52-
->getDefinition($serviceId)
53-
->replaceArgument(0, new Reference($managerRegistry));
5483
}
5584

56-
/**
57-
* @param string $driver
58-
*
59-
* @return string
60-
*/
6185
private function getManagerRegistry(string $driver): string
6286
{
6387
switch ($driver) {
@@ -70,11 +94,6 @@ private function getManagerRegistry(string $driver): string
7094
}
7195
}
7296

73-
/**
74-
* @param string $driver
75-
*
76-
* @return string
77-
*/
7897
private function getEventSubscriber(string $driver): string
7998
{
8099
switch ($driver) {

‎Doctrine/Queue/SyncIndexWithObjectChangeProcessor.php

+13-31
Original file line numberDiff line numberDiff line change
@@ -7,32 +7,23 @@
77
use Enqueue\Util\JSON;
88
use FOS\ElasticaBundle\Persister\PersisterRegistry;
99
use FOS\ElasticaBundle\Provider\IndexableInterface;
10-
use Interop\Queue\PsrContext;
11-
use Interop\Queue\PsrMessage;
12-
use Interop\Queue\PsrProcessor;
10+
use Interop\Queue\Context;
11+
use Interop\Queue\Message;
12+
use Interop\Queue\Processor;
1313
use Doctrine\Common\Persistence\ManagerRegistry;
1414

15-
final class SyncIndexWithObjectChangeProcessor implements PsrProcessor, CommandSubscriberInterface, QueueSubscriberInterface
15+
final class SyncIndexWithObjectChangeProcessor implements Processor, CommandSubscriberInterface, QueueSubscriberInterface
1616
{
1717
const INSERT_ACTION = 'insert';
1818

1919
const UPDATE_ACTION = 'update';
2020

2121
const REMOVE_ACTION = 'remove';
2222

23-
/**
24-
* @var PersisterRegistry
25-
*/
2623
private $persisterRegistry;
2724

28-
/**
29-
* @var IndexableInterface
30-
*/
3125
private $indexable;
3226

33-
/**
34-
* @var ManagerRegistry
35-
*/
3627
private $doctrine;
3728

3829
public function __construct(ManagerRegistry $doctrine, PersisterRegistry $persisterRegistry, IndexableInterface $indexable)
@@ -42,10 +33,7 @@ public function __construct(ManagerRegistry $doctrine, PersisterRegistry $persis
4233
$this->doctrine = $doctrine;
4334
}
4435

45-
/**
46-
* {@inheritdoc}
47-
*/
48-
public function process(PsrMessage $message, PsrContext $context)
36+
public function process(Message $message, Context $context): Result
4937
{
5038
$data = JSON::decode($message->getBody());
5139

@@ -94,7 +82,7 @@ public function process(PsrMessage $message, PsrContext $context)
9482
}
9583
}
9684

97-
return self::ACK;
85+
return Result::ack();
9886
case self::INSERT_ACTION:
9987
if (false == $object = $repository->{$repositoryMethod}($id)) {
10088
$persister->deleteById($id);
@@ -106,33 +94,27 @@ public function process(PsrMessage $message, PsrContext $context)
10694
$persister->insertOne($object);
10795
}
10896

109-
return self::ACK;
97+
return Result::ack();
11098
case self::REMOVE_ACTION:
11199
$persister->deleteById($id);
112100

113-
return self::ACK;
101+
return Result::ack();
114102
default:
115103
return Result::reject(sprintf('The action "%s" is not supported', $action));
116104
}
117105
}
118106

119-
/**
120-
* {@inheritdoc}
121-
*/
122-
public static function getSubscribedCommand()
107+
public static function getSubscribedCommand(): array
123108
{
124109
return [
125-
'processorName' => Commands::SYNC_INDEX_WITH_OBJECT_CHANGE,
126-
'queueName' => Commands::SYNC_INDEX_WITH_OBJECT_CHANGE,
127-
'queueNameHardcoded' => true,
110+
'command' => Commands::SYNC_INDEX_WITH_OBJECT_CHANGE,
111+
'queue' => Commands::SYNC_INDEX_WITH_OBJECT_CHANGE,
112+
'prefix_queue' => false,
128113
'exclusive' => true,
129114
];
130115
}
131116

132-
/**
133-
* {@inheritdoc}
134-
*/
135-
public static function getSubscribedQueues()
117+
public static function getSubscribedQueues(): array
136118
{
137119
return [Commands::SYNC_INDEX_WITH_OBJECT_CHANGE];
138120
}

‎Doctrine/SyncIndexWithObjectChangeListener.php

+2-5
Original file line numberDiff line numberDiff line change
@@ -5,14 +5,11 @@
55
use Enqueue\ElasticaBundle\Doctrine\Queue\Commands;
66
use Enqueue\ElasticaBundle\Doctrine\Queue\SyncIndexWithObjectChangeProcessor as SyncProcessor;
77
use Enqueue\Util\JSON;
8-
use Interop\Queue\PsrContext;
8+
use Interop\Queue\Context;
99
use Doctrine\Common\EventSubscriber;
1010

1111
final class SyncIndexWithObjectChangeListener implements EventSubscriber
1212
{
13-
/**
14-
* @var PsrContext
15-
*/
1613
private $context;
1714

1815
/**
@@ -25,7 +22,7 @@ final class SyncIndexWithObjectChangeListener implements EventSubscriber
2522
*/
2623
private $config;
2724

28-
public function __construct(PsrContext $context, $modelClass, array $config)
25+
public function __construct(Context $context, $modelClass, array $config)
2926
{
3027
$this->context = $context;
3128
$this->modelClass = $modelClass;

‎Persister/Listener/PurgePopulateQueueListener.php

+2-8
Original file line numberDiff line numberDiff line change
@@ -2,21 +2,15 @@
22
namespace Enqueue\ElasticaBundle\Persister\Listener;
33

44
use FOS\ElasticaBundle\Persister\Event\PrePersistEvent;
5-
use Interop\Queue\PsrContext;
5+
use Interop\Queue\Context;
66
use FOS\ElasticaBundle\Persister\Event\Events;
77
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
88

99
class PurgePopulateQueueListener implements EventSubscriberInterface
1010
{
11-
/**
12-
* @var PsrContext
13-
*/
1411
private $context;
1512

16-
/**
17-
* @param PsrContext $context
18-
*/
19-
public function __construct(PsrContext $context)
13+
public function __construct(Context $context)
2014
{
2115
$this->context = $context;
2216
}

‎Persister/QueuePagerPersister.php

+2-5
Original file line numberDiff line numberDiff line change
@@ -10,16 +10,13 @@
1010
use FOS\ElasticaBundle\Persister\PagerPersisterInterface;
1111
use FOS\ElasticaBundle\Persister\PersisterRegistry;
1212
use FOS\ElasticaBundle\Provider\PagerInterface;
13-
use Interop\Queue\PsrContext;
13+
use Interop\Queue\Context;
1414
use Symfony\Component\EventDispatcher\EventDispatcherInterface;
1515

1616
final class QueuePagerPersister implements PagerPersisterInterface
1717
{
1818
const NAME = 'queue';
1919

20-
/**
21-
* @var PsrContext
22-
*/
2320
private $context;
2421

2522
/**
@@ -32,7 +29,7 @@ final class QueuePagerPersister implements PagerPersisterInterface
3229
*/
3330
private $dispatcher;
3431

35-
public function __construct(PsrContext $context, PersisterRegistry $registry, EventDispatcherInterface $dispatcher)
32+
public function __construct(Context $context, PersisterRegistry $registry, EventDispatcherInterface $dispatcher)
3633
{
3734
$this->context = $context;
3835
$this->dispatcher = $dispatcher;

‎Queue/PopulateProcessor.php

+12-42
Original file line numberDiff line numberDiff line change
@@ -7,21 +7,15 @@
77
use FOS\ElasticaBundle\Persister\InPlacePagerPersister;
88
use FOS\ElasticaBundle\Persister\PagerPersisterRegistry;
99
use FOS\ElasticaBundle\Provider\PagerProviderRegistry;
10-
use Interop\Queue\PsrContext;
11-
use Interop\Queue\PsrMessage;
12-
use Interop\Queue\PsrProcessor;
10+
use Interop\Queue\Context;
11+
use Interop\Queue\Message;
12+
use Interop\Queue\Processor;
1313
use Enqueue\Util\JSON;
1414

15-
final class PopulateProcessor implements PsrProcessor, CommandSubscriberInterface, QueueSubscriberInterface
15+
final class PopulateProcessor implements Processor, CommandSubscriberInterface, QueueSubscriberInterface
1616
{
17-
/**
18-
* @var PagerProviderRegistry
19-
*/
2017
private $pagerProviderRegistry;
2118

22-
/**
23-
* @var PagerPersisterRegistry
24-
*/
2519
private $pagerPersisterRegistry;
2620

2721
public function __construct(
@@ -32,10 +26,7 @@ public function __construct(
3226
$this->pagerProviderRegistry = $pagerProviderRegistry;
3327
}
3428

35-
/**
36-
* {@inheritdoc}
37-
*/
38-
public function process(PsrMessage $message, PsrContext $context)
29+
public function process(Message $message, Context $context): Result
3930
{
4031
if ($message->isRedelivered()) {
4132
$replyMessage = $this->createReplyMessage($context, $message, 0,'The message was redelivered. Chances are that something has gone wrong.');
@@ -81,14 +72,7 @@ public function process(PsrMessage $message, PsrContext $context)
8172
}
8273
}
8374

84-
/**
85-
* @param PsrContext $context
86-
* @param PsrMessage $message
87-
* @param int $objectsCount
88-
* @param \Throwable $e
89-
* @return PsrMessage
90-
*/
91-
private function createExceptionReplyMessage(PsrContext $context, PsrMessage $message, $objectsCount, \Throwable $e)
75+
private function createExceptionReplyMessage(Context $context, Message $message, int $objectsCount, \Throwable $e): Message
9276
{
9377
$errorMessage = sprintf(
9478
'<error>The queue processor has failed to process the message with exception: </error><comment>%s: %s in file %s at line %s.</comment>',
@@ -101,15 +85,7 @@ private function createExceptionReplyMessage(PsrContext $context, PsrMessage $me
10185
return $this->createReplyMessage($context, $message, $errorMessage);
10286
}
10387

104-
/**
105-
* @param PsrContext $context
106-
* @param PsrMessage $message
107-
* @param int $objectsCount
108-
* @param string|null $error
109-
*
110-
* @return PsrMessage
111-
*/
112-
private function createReplyMessage(PsrContext $context, PsrMessage $message, $objectsCount, $error = null)
88+
private function createReplyMessage(Context $context, Message $message, int $objectsCount, string $error = null): Message
11389
{
11490
$replyMessage = $context->createMessage($message->getBody(), $message->getProperties(), $message->getHeaders());
11591
$replyMessage->setProperty('fos-populate-objects-count', $objectsCount);
@@ -121,23 +97,17 @@ private function createReplyMessage(PsrContext $context, PsrMessage $message, $o
12197
return $replyMessage;
12298
}
12399

124-
/**
125-
* {@inheritdoc}
126-
*/
127-
public static function getSubscribedCommand()
100+
public static function getSubscribedCommand(): array
128101
{
129102
return [
130-
'processorName' => Commands::POPULATE,
131-
'queueName' => Commands::POPULATE,
132-
'queueNameHardcoded' => true,
103+
'command' => Commands::POPULATE,
104+
'queue' => Commands::POPULATE,
105+
'prefix_queue' => false,
133106
'exclusive' => true,
134107
];
135108
}
136109

137-
/**
138-
* {@inheritdoc}
139-
*/
140-
public static function getSubscribedQueues()
110+
public static function getSubscribedQueues(): array
141111
{
142112
return [Commands::POPULATE];
143113
}

0 commit comments

Comments
 (0)