Skip to content

Commit 14e0fd8

Browse files
makasimmakasim
committed
[client] Redis driver
1 parent 6b0efaf commit 14e0fd8

10 files changed

+941
-2
lines changed

‎Client/RedisDriver.php

+151
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,151 @@
1+
<?php
2+
3+
namespace Enqueue\Redis\Client;
4+
5+
use Enqueue\Client\Config;
6+
use Enqueue\Client\DriverInterface;
7+
use Enqueue\Client\Message;
8+
use Enqueue\Client\MessagePriority;
9+
use Enqueue\Client\Meta\QueueMetaRegistry;
10+
use Enqueue\Psr\PsrMessage;
11+
use Enqueue\Redis\RedisContext;
12+
use Enqueue\Redis\RedisDestination;
13+
use Enqueue\Redis\RedisMessage;
14+
use Psr\Log\LoggerInterface;
15+
use Psr\Log\NullLogger;
16+
17+
class RedisDriver implements DriverInterface
18+
{
19+
/**
20+
* @var RedisContext
21+
*/
22+
private $context;
23+
24+
/**
25+
* @var Config
26+
*/
27+
private $config;
28+
29+
/**
30+
* @var QueueMetaRegistry
31+
*/
32+
private $queueMetaRegistry;
33+
34+
/**
35+
* @param RedisContext $context
36+
* @param Config $config
37+
* @param QueueMetaRegistry $queueMetaRegistry
38+
*/
39+
public function __construct(RedisContext $context, Config $config, QueueMetaRegistry $queueMetaRegistry)
40+
{
41+
$this->context = $context;
42+
$this->config = $config;
43+
$this->queueMetaRegistry = $queueMetaRegistry;
44+
}
45+
46+
/**
47+
* {@inheritdoc}
48+
*/
49+
public function sendToRouter(Message $message)
50+
{
51+
if (false == $message->getProperty(Config::PARAMETER_TOPIC_NAME)) {
52+
throw new \LogicException('Topic name parameter is required but is not set');
53+
}
54+
55+
$queue = $this->createQueue($this->config->getRouterQueueName());
56+
$transportMessage = $this->createTransportMessage($message);
57+
58+
$this->context->createProducer()->send($queue, $transportMessage);
59+
}
60+
61+
/**
62+
* {@inheritdoc}
63+
*/
64+
public function sendToProcessor(Message $message)
65+
{
66+
if (false == $message->getProperty(Config::PARAMETER_PROCESSOR_NAME)) {
67+
throw new \LogicException('Processor name parameter is required but is not set');
68+
}
69+
70+
if (false == $queueName = $message->getProperty(Config::PARAMETER_PROCESSOR_QUEUE_NAME)) {
71+
throw new \LogicException('Queue name parameter is required but is not set');
72+
}
73+
74+
$transportMessage = $this->createTransportMessage($message);
75+
$destination = $this->createQueue($queueName);
76+
77+
$this->context->createProducer()->send($destination, $transportMessage);
78+
}
79+
80+
/**
81+
* {@inheritdoc}
82+
*/
83+
public function setupBroker(LoggerInterface $logger = null)
84+
{
85+
}
86+
87+
/**
88+
* {@inheritdoc}
89+
*
90+
* @return RedisDestination
91+
*/
92+
public function createQueue($queueName)
93+
{
94+
return $this->context->createQueue($this->config->createTransportQueueName($queueName));
95+
}
96+
97+
/**
98+
* {@inheritdoc}
99+
*
100+
* @return RedisMessage
101+
*/
102+
public function createTransportMessage(Message $message)
103+
{
104+
$properties = $message->getProperties();
105+
106+
$headers = $message->getHeaders();
107+
$headers['content_type'] = $message->getContentType();
108+
109+
$transportMessage = $this->context->createMessage();
110+
$transportMessage->setBody($message->getBody());
111+
$transportMessage->setHeaders($headers);
112+
$transportMessage->setProperties($properties);
113+
$transportMessage->setMessageId($message->getMessageId());
114+
$transportMessage->setTimestamp($message->getTimestamp());
115+
$transportMessage->setReplyTo($message->getReplyTo());
116+
$transportMessage->setCorrelationId($message->getCorrelationId());
117+
118+
return $transportMessage;
119+
}
120+
121+
/**
122+
* @param RedisMessage $message
123+
*
124+
* {@inheritdoc}
125+
*/
126+
public function createClientMessage(PsrMessage $message)
127+
{
128+
$clientMessage = new Message();
129+
130+
$clientMessage->setBody($message->getBody());
131+
$clientMessage->setHeaders($message->getHeaders());
132+
$clientMessage->setProperties($message->getProperties());
133+
134+
$clientMessage->setContentType($message->getHeader('content_type'));
135+
$clientMessage->setMessageId($message->getMessageId());
136+
$clientMessage->setTimestamp($message->getTimestamp());
137+
$clientMessage->setPriority(MessagePriority::NORMAL);
138+
$clientMessage->setReplyTo($message->getReplyTo());
139+
$clientMessage->setCorrelationId($message->getCorrelationId());
140+
141+
return $clientMessage;
142+
}
143+
144+
/**
145+
* @return Config
146+
*/
147+
public function getConfig()
148+
{
149+
return $this->config;
150+
}
151+
}

‎PRedis.php

+1-1
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@ public function lpush($key, $value)
3737
public function brpop($key, $timeout)
3838
{
3939
try {
40-
if ($result = $this->redis->brpop($key, $timeout)) {
40+
if ($result = $this->redis->brpop([$key], $timeout)) {
4141
return $result[1];
4242
}
4343
} catch (PRedisServerException $e) {

‎RedisConsumer.php

+10-1
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,16 @@ public function getQueue()
4545
*/
4646
public function receive($timeout = 0)
4747
{
48-
if ($message = $this->getRedis()->brpop($this->queue->getName(), (int) $timeout / 1000)) {
48+
$timeout = (int) ($timeout / 1000);
49+
if (empty($timeout)) {
50+
// Caused by
51+
// Predis\Response\ServerException: ERR timeout is not an integer or out of range
52+
// /mqdev/vendor/predis/predis/src/Client.php:370
53+
54+
return $this->receiveNoWait();
55+
}
56+
57+
if ($message = $this->getRedis()->brpop($this->queue->getName(), $timeout)) {
4958
return RedisMessage::jsonUnserialize($message);
5059
}
5160
}

‎Symfony/RedisTransportFactory.php

+114
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,114 @@
1+
<?php
2+
3+
namespace Enqueue\Redis\Symfony;
4+
5+
use Enqueue\Redis\Client\RedisDriver;
6+
use Enqueue\Redis\RedisConnectionFactory;
7+
use Enqueue\Redis\RedisContext;
8+
use Enqueue\Symfony\TransportFactoryInterface;
9+
use Symfony\Component\Config\Definition\Builder\ArrayNodeDefinition;
10+
use Symfony\Component\DependencyInjection\ContainerBuilder;
11+
use Symfony\Component\DependencyInjection\Definition;
12+
use Symfony\Component\DependencyInjection\Reference;
13+
14+
class RedisTransportFactory implements TransportFactoryInterface
15+
{
16+
/**
17+
* @var string
18+
*/
19+
private $name;
20+
21+
/**
22+
* @param string $name
23+
*/
24+
public function __construct($name = 'redis')
25+
{
26+
$this->name = $name;
27+
}
28+
29+
/**
30+
* {@inheritdoc}
31+
*/
32+
public function addConfiguration(ArrayNodeDefinition $builder)
33+
{
34+
$builder
35+
->children()
36+
->scalarNode('host')
37+
->isRequired()
38+
->cannotBeEmpty()
39+
->info('can be a host, or the path to a unix domain socket')
40+
->end()
41+
->integerNode('port')->end()
42+
->enumNode('vendor')
43+
->values(['phpredis', 'predis'])
44+
->isRequired()
45+
->cannotBeEmpty()
46+
->info('The library used internally to interact with Redis server')
47+
->end()
48+
->booleanNode('persisted')
49+
->defaultFalse()
50+
->info('bool, Whether it use single persisted connection or open a new one for every context')
51+
->end()
52+
->booleanNode('lazy')
53+
->defaultTrue()
54+
->info('the connection will be performed as later as possible, if the option set to true')
55+
->end()
56+
;
57+
}
58+
59+
/**
60+
* {@inheritdoc}
61+
*/
62+
public function createConnectionFactory(ContainerBuilder $container, array $config)
63+
{
64+
$factory = new Definition(RedisConnectionFactory::class);
65+
$factory->setArguments([$config]);
66+
67+
$factoryId = sprintf('enqueue.transport.%s.connection_factory', $this->getName());
68+
$container->setDefinition($factoryId, $factory);
69+
70+
return $factoryId;
71+
}
72+
73+
/**
74+
* {@inheritdoc}
75+
*/
76+
public function createContext(ContainerBuilder $container, array $config)
77+
{
78+
$factoryId = sprintf('enqueue.transport.%s.connection_factory', $this->getName());
79+
80+
$context = new Definition(RedisContext::class);
81+
$context->setFactory([new Reference($factoryId), 'createContext']);
82+
83+
$contextId = sprintf('enqueue.transport.%s.context', $this->getName());
84+
$container->setDefinition($contextId, $context);
85+
86+
return $contextId;
87+
}
88+
89+
/**
90+
* {@inheritdoc}
91+
*/
92+
public function createDriver(ContainerBuilder $container, array $config)
93+
{
94+
$driver = new Definition(RedisDriver::class);
95+
$driver->setArguments([
96+
new Reference(sprintf('enqueue.transport.%s.context', $this->getName())),
97+
new Reference('enqueue.client.config'),
98+
new Reference('enqueue.client.meta.queue_meta_registry'),
99+
]);
100+
101+
$driverId = sprintf('enqueue.client.%s.driver', $this->getName());
102+
$container->setDefinition($driverId, $driver);
103+
104+
return $driverId;
105+
}
106+
107+
/**
108+
* {@inheritdoc}
109+
*/
110+
public function getName()
111+
{
112+
return $this->name;
113+
}
114+
}

0 commit comments

Comments
 (0)