forked from semji/queue-client-bundle
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathQueueClientFactory.php
More file actions
72 lines (65 loc) · 2.59 KB
/
QueueClientFactory.php
File metadata and controls
72 lines (65 loc) · 2.59 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
<?php
namespace ReputationVIP\Bundle\QueueClientBundle;
use Psr\Log\LoggerInterface;
use ReputationVIP\Bundle\QueueClientBundle\Configuration\QueuesConfiguration;
use ReputationVIP\QueueClient\Adapter\AdapterInterface;
use ReputationVIP\QueueClient\Adapter\Exception\QueueAccessException;
use ReputationVIP\QueueClient\QueueClient;
use ReputationVIP\QueueClient\QueueClientInterface;
use Symfony\Component\Config\Definition\Processor;
use Symfony\Component\DependencyInjection\ContainerInterface;
use Symfony\Component\Yaml\Yaml;
class QueueClientFactory
{
/**
* @param $item
* @param $key
* @param ContainerInterface $container
*/
public static function resolveParameters(&$item, $key, $container)
{
if (!is_array($item)) {
$matches = [];
while (preg_match('/(?<=%)(.*?)(?=%)/', $item, $matches)) {
$param = $matches[0];
if (!empty($param)) {
$item = str_replace('%' . $param . '%', $container->getParameter($param), $item);
} else {
throw new \InvalidArgumentException('Empty parameter!');
}
}
}
}
/**
* @param ContainerInterface $container
* @param AdapterInterface $adapter
* @param string $queuesFile
* @return null|QueueClientInterface
* @throws \ErrorException
*/
public function get($container, $adapter, $queuesFile)
{
/** @var LoggerInterface $logger */
$logger = $container->get('logger');
$queueClient = new QueueClient($adapter);
$processor = new Processor();
$configuration = new QueuesConfiguration();
$processedConfiguration = $processor->processConfiguration($configuration, Yaml::parse(file_get_contents($queuesFile)));
array_walk_recursive($processedConfiguration, 'ReputationVIP\Bundle\QueueClientBundle\QueueClientFactory::resolveParameters', $container);
foreach ($processedConfiguration[QueuesConfiguration::QUEUES_NODE] as $queue) {
$queueName = $queue[QueuesConfiguration::QUEUE_NAME_NODE];
foreach ($queue[QueuesConfiguration::QUEUE_ALIASES_NODE] as $alias) {
try {
$queueClient->addAlias($queueName, $alias);
} catch (QueueAccessException $e) {
$logger->warning($e->getMessage());
} catch (\ErrorException $e) {
if ($e->getSeverity() === E_ERROR) {
throw $e;
}
}
}
}
return $queueClient;
}
}