8000 [Messenger] Fix missing auto_setup for RedisTransport · symfony/symfony@d27bc2a · GitHub
[go: up one dir, main page]
Skip to content

Commit d27bc2a

Browse files
author
Robin Chalas
committed
[Messenger] Fix missing auto_setup for RedisTransport
1 parent 6aacfea commit d27bc2a

File tree

2 files changed

+34
-17
lines changed

2 files changed

+34
-17
lines changed

‎src/Symfony/Component/Messenger/Tests/Transport/RedisExt/ConnectionTest.php

Lines changed: 2 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -42,13 +42,13 @@ public function testFromDsn()
4242
public function testFromDsnWithOptions()
4343
{
4444
$this->assertEquals(
45-
new Connection(['stream' => 'queue', 'group' => 'group1', 'consumer' => 'consumer1'], [
45+
new Connection(['stream' => 'queue', 'group' => 'group1', 'consumer' => 'consumer1', 'auto_setup' => false], [
4646
'host' => 'localhost',
4747
'port' => 6379,
4848
], [
4949
'serializer' => 2,
5050
]),
51-
Connection::fromDsn('redis://localhost/queue/group1/consumer1', ['serializer' => 2])
51+
Connection::fromDsn('redis://localhost/queue/group1/consumer1', ['serializer' => 2, 'auto_setup' => false])
5252
);
5353
}
5454

@@ -117,10 +117,6 @@ public function testGetAfterReject()
117117
{
118118
$redis = new \Redis();
119119
$connection = Connection::fromDsn('redis://localhost/messenger-rejectthenget', [], $redis);
120-
try {
121-
$connection->setup();
122-
} catch (TransportException $e) {
123-
}
124120

125121
$connection->add('1', []);
126122
$connection->add('2', []);
@@ -139,10 +135,6 @@ public function testGetNonBlocking()
139135
$redis = new \Redis();
140136

141137
$connection = Connection::fromDsn('redis://localhost/messenger-getnonblocking', [], $redis);
142-
try {
143-
$connection->setup();
144-
} catch (TransportException $e) {
145-
}
146138

147139
$this->assertNull($connection->get()); // no message, should return null immediately
148140
$connection->add('1', []);

‎src/Symfony/Component/Messenger/Transport/RedisExt/Connection.php

Lines changed: 32 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -27,20 +27,29 @@
2727
*/
2828
class Connection
2929
{
30+
private const DEFAULT_OPTIONS = [
31+
'stream' => 'messages',
32+
'group' => 'symfony',
33+
'consumer' => 'consumer',
34+
'auto_setup' => true,
35+
];
36+
3037
private $connection;
3138
private $stream;
3239
private $group;
3340
private $consumer;
41+
private $autoSetup;
3442
private $couldHavePendingMessages = true;
3543

3644
public function __construct(array $configuration, array $connectionCredentials = [], array $redisOptions = [], \Redis $redis = null)
3745
{
3846
$this->connection = $redis ?: new \Redis();
3947
$this->connection->connect($connectionCredentials['host'] ?? '127.0.0.1', $connectionCredentials['port'] ?? 6379);
4048
$this->connection->setOption(\Redis::OPT_SERIALIZER, $redisOptions['serializer'] ?? \Redis::SERIALIZER_PHP);
41-
$this->stream = $configuration['stream'] ?? '' ?: 'messages';
42-
$this->group = $configuration['group'] ?? '' ?: 'symfony';
43-
$this->consumer = $configuration['consumer'] ?? '' ?: 'consumer';
49+
$this->stream = $configuration['stream'] ?? self::DEFAULT_OPTIONS['stream'];
50+
$this->group = $configuration['group'] ?? self::DEFAULT_OPTIONS['group'];
51+
$this->consumer = $configuration['consumer'] ?? self::DEFAULT_OPTIONS['consumer'];
52+
$this->autoSetup = $configuration['auto_setup'] ?? self::DEFAULT_OPTIONS['auto_setup'];
4453
}
4554

4655
public static function fromDsn(string $dsn, array $redisOptions = [], \Redis $redis = null): self
@@ -51,9 +60,9 @@ public static function fromDsn(string $dsn, array $redisOptions = [], \Redis $re
5160

5261
$pathParts = explode('/', $parsedUrl['path'] ?? '');
5362

54-
$stream = $pathParts[1] ?? '';
55-
$group = $pathParts[2] ?? '';
56-
$consumer = $pathParts[3] ?? '';
63+
$stream = $pathParts[1] ?? null;
64+
$group = $pathParts[2] ?? null;
65+
$consumer = $pathParts[3] ?? null;
5766

5867
$connectionCredentials = [
5968
'host' => $parsedUrl['host'] ?? '127.0.0.1',
@@ -64,11 +73,21 @@ public static function fromDsn(string $dsn, array $redisOptions = [], \Redis $re
6473
parse_str($parsedUrl['query'], $redisOptions);
6574
}
6675

67-
return new self(['stream' => $stream, 'group' => $group, 'consumer' => $consumer], $connectionCredentials, $redisOptions, $redis);
76+
$autoSetup = null;
77+
if (\array_key_exists('auto_setup', $redisOptions)) {
78+
$autoSetup = filter_var($redisOptions['auto_setup'], FILTER_VALIDATE_BOOLEAN);
79+
unset($redisOptions['auto_setup']);
80+
}
81+
82+
return new self(['stream' => $stream, 'group' => $group, 'consumer' => $consumer, 'auto_setup' => $autoSetup], $connectionCredentials, $redisOptions, $redis);
6883
}
6984

7085
public function get(): ?array
7186
{
87+
if ($this->autoSetup) {
88+
$this->setup();
89+
}
90+
7291
$messageId = '>'; // will receive new messages
7392

7493
if ($this->couldHavePendingMessages) {
@@ -141,6 +160,10 @@ public function reject(string $id): void
141160

142161
public function add(string $body, array $headers): void
143162
{
163+
if ($this->autoSetup) {
164+
$this->setup();
165+
}
166+
144167
$e = null;
145168
try {
146169
$added = $this->connection->xadd($this->stream, '*', ['message' => json_encode(
@@ -161,5 +184,7 @@ public function setup(): void
161184
} catch (\RedisException $e) {
162185
throw new TransportException($e->getMessage(), 0, $e);
163186
}
187+
188+
$this->autoSetup = false;
164189
}
165190
}

0 commit comments

Comments
 (0)
0