Skip to content

Commit 2707e9c

Browse files
taylorotwelltaylorotwell
committed
Merge branch 'feature/redis-cluster' into 13.x
2 parents 1741307 + 64a0ff2 commit 2707e9c

11 files changed

Lines changed: 846 additions & 90 deletions

File tree

‎.github/workflows/redis.yml‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ on:
44
push:
55
branches:
66
- master
7-
- '*.x'
7+
- "*.x"
88
pull_request:
99

1010
jobs:
@@ -21,7 +21,7 @@ jobs:
2121
strategy:
2222
fail-fast: true
2323
matrix:
24-
client: ['phpredis', 'predis']
24+
client: ["phpredis", "predis"]
2525

2626
name: Redis (${{ matrix.client}}) Driver
2727

@@ -67,7 +67,7 @@ jobs:
6767
strategy:
6868
fail-fast: true
6969
matrix:
70-
client: ['phpredis', 'predis']
70+
client: ["phpredis", "predis"]
7171

7272
name: Redis Cluster (${{ matrix.client}}) Driver
7373

@@ -129,6 +129,5 @@ jobs:
129129
env:
130130
REDIS_CLIENT: ${{ matrix.client }}
131131
REDIS_CLUSTER_HOSTS_AND_PORTS: 127.0.0.1:7000,127.0.0.1:7001,127.0.0.1:7002
132-
REDIS_QUEUE: '{default}'
132+
REDIS_QUEUE: "default"
133133
QUEUE_CONNECTION: redis
134-

‎src/Illuminate/Queue/RedisQueue.php‎

Lines changed: 50 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
use Illuminate\Contracts\Redis\Factory as Redis;
88
use Illuminate\Queue\Jobs\InspectedJob;
99
use Illuminate\Queue\Jobs\RedisJob;
10+
use Illuminate\Redis\Connections\Connection;
1011
use Illuminate\Redis\Connections\PhpRedisClusterConnection;
1112
use Illuminate\Redis\Connections\PredisClusterConnection;
1213
use Illuminate\Support\Collection;
@@ -67,6 +68,13 @@ class RedisQueue extends Queue implements QueueContract, ClearableQueue
6768
*/
6869
protected $secondaryQueueHadJob = false;
6970

71+
/**
72+
* Indicates if the connection is a Redis Cluster connection.
73+
*
74+
* @var bool|null
75+
*/
76+
protected $isCluster = null;
77+
7078
/**
7179
* Create a new Redis queue instance.
7280
*
@@ -104,7 +112,7 @@ public function __construct(
104112
*/
105113
public function size($queue = null)
106114
{
107-
$queue = $this->getQueue($queue);
115+
$queue = $this->getQueueRedisKey($queue);
108116

109117
return $this->getConnection()->eval(
110118
LuaScripts::size(), 3, $queue, $queue.':delayed', $queue.':reserved'
@@ -119,7 +127,7 @@ public function size($queue = null)
119127
*/
120128
public function pendingSize($queue = null)
121129
{
122-
return $this->getConnection()->llen($this->getQueue($queue));
130+
return $this->getConnection()->llen($this->getQueueRedisKey($queue));
123131
}
124132

125133
/**
@@ -130,7 +138,7 @@ public function pendingSize($queue = null)
130138
*/
131139
public function delayedSize($queue = null)
132140
{
133-
return $this->getConnection()->zcard($this->getQueue($queue).':delayed');
141+
return $this->getConnection()->zcard($this->getQueueRedisKey($queue).':delayed');
134142
}
135143

136144
/**
@@ -141,7 +149,7 @@ public function delayedSize($queue = null)
141149
*/
142150
public function reservedSize($queue = null)
143151
{
144-
return $this->getConnection()->zcard($this->getQueue($queue).':reserved');
152+
return $this->getConnection()->zcard($this->getQueueRedisKey($queue).':reserved');
145153
}
146154

147155
/**
@@ -152,7 +160,7 @@ public function reservedSize($queue = null)
152160
*/
153161
public function pendingJobs($queue = null): Collection
154162
{
155-
$queue = $this->getQueue($queue);
163+
$queue = $this->getQueueRedisKey($queue);
156164

157165
return (new Collection($this->getConnection()->lrange($queue, 0, -1)))
158166
->map(fn ($payload) => InspectedJob::fromPayload($payload));
@@ -166,7 +174,7 @@ public function pendingJobs($queue = null): Collection
166174
*/
167175
public function delayedJobs($queue = null): Collection
168176
{
169-
$queue = $this->getQueue($queue);
177+
$queue = $this->getQueueRedisKey($queue);
170178

171179
return (new Collection($this->getConnection()->zrange($queue.':delayed', 0, -1)))
172180
->map(fn ($payload) => InspectedJob::fromPayload($payload));
@@ -180,7 +188,7 @@ public function delayedJobs($queue = null): Collection
180188
*/
181189
public function reservedJobs($queue = null): Collection
182190
{
183-
$queue = $this->getQueue($queue);
191+
$queue = $this->getQueueRedisKey($queue);
184192

185193
return (new Collection($this->getConnection()->zrange($queue.':reserved', 0, -1)))
186194
->map(fn ($payload) => InspectedJob::fromPayload($payload));
@@ -194,7 +202,7 @@ public function reservedJobs($queue = null): Collection
194202
*/
195203
public function creationTimeOfOldestPendingJob($queue = null)
196204
{
197-
$payload = $this->getConnection()->lindex($this->getQueue($queue), 0);
205+
$payload = $this->getConnection()->lindex($this->getQueueRedisKey($queue), 0);
198206

199207
if (! $payload) {
200208
return null;
@@ -267,9 +275,11 @@ function ($payload, $queue) {
267275
*/
268276
public function pushRaw($payload, $queue = null, array $options = [])
269277
{
278+
$queue = $this->getQueueRedisKey($queue);
279+
270280
$this->getConnection()->eval(
271-
LuaScripts::push(), 2, $this->getQueue($queue),
272-
$this->getQueue($queue).':notify', $payload
281+
LuaScripts::push(), 2, $queue,
282+
$queue.':notify', $payload
273283
);
274284

275285
return json_decode($payload, true)['id'] ?? null;
@@ -308,7 +318,7 @@ function ($payload, $queue, $delay) {
308318
protected function laterRaw($delay, $payload, $queue = null)
309319
{
310320
$this->getConnection()->eval(
311-
LuaScripts::later(), 1, $this->getQueue($queue).':delayed',
321+
LuaScripts::later(), 1, $this->getQueueRedisKey($queue).':delayed',
312322
$this->availableAt($delay), $payload
313323
);
314324

@@ -340,7 +350,7 @@ protected function createPayloadArray($job, $queue, $data = '')
340350
*/
341351
public function pop($queue = null, $index = 0)
342352
{
343-
$this->migrate($prefixed = $this->getQueue($queue));
353+
$this->migrate($prefixed = $this->getQueueRedisKey($queue));
344354

345355
$block = ! $this->secondaryQueueHadJob && $index == 0;
346356

@@ -428,7 +438,7 @@ protected function retrieveNextJob($queue, $block = true)
428438
*/
429439
public function deleteReserved($queue, $job)
430440
{
431-
$this->getConnection()->zrem($this->getQueue($queue).':reserved', $job->getReservedJob());
441+
$this->getConnection()->zrem($this->getQueueRedisKey($queue).':reserved', $job->getReservedJob());
432442
}
433443

434444
/**
@@ -441,7 +451,7 @@ public function deleteReserved($queue, $job)
441451
*/
442452
public function deleteAndRelease($queue, $job, $delay)
443453
{
444-
$queue = $this->getQueue($queue);
454+
$queue = $this->getQueueRedisKey($queue);
445455

446456
$this->getConnection()->eval(
447457
LuaScripts::release(), 2, $queue.':delayed', $queue.':reserved',
@@ -457,7 +467,7 @@ public function deleteAndRelease($queue, $job, $delay)
457467
*/
458468
public function clear($queue)
459469
{
460-
$queue = $this->getQueue($queue);
470+
$queue = $this->getQueueRedisKey($queue);
461471

462472
return $this->getConnection()->eval(
463473
LuaScripts::clear(), 4, $queue, $queue.':delayed',
@@ -486,6 +496,21 @@ public function getQueue($queue)
486496
return 'queues:'.($queue ?: $this->default);
487497
}
488498

499+
/**
500+
* Get the cluster-safe Redis key for the given queue.
501+
*
502+
* @param string|null $queue
503+
* @return string
504+
*/
505+
protected function getQueueRedisKey($queue = null)
506+
{
507+
$queue = $queue ?: $this->default;
508+
509+
return $this->isClusterConnection() && ! Connection::hasHashTag($queue)
510+
? $this->getQueue('{'.$queue.'}')
511+
: $this->getQueue($queue);
512+
}
513+
489514
/**
490515
* Get the connection for the queue.
491516
*
@@ -496,6 +521,16 @@ public function getConnection()
496521
return $this->redis->connection($this->connection);
497522
}
498523

524+
/**
525+
* Determine if the connection is a Redis Cluster connection.
526+
*
527+
* @return bool
528+
*/
529+
protected function isClusterConnection()
530+
{
531+
return $this->isCluster ??= $this->getConnection()->isCluster();
532+
}
533+
499534
/**
500535
* Get the underlying Redis instance.
501536
*

‎src/Illuminate/Redis/Connections/Connection.php‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -182,6 +182,16 @@ public function listenForFailures(Closure $callback)
182182
$this->events?->listen(CommandFailed::class, $callback);
183183
}
184184

185+
/**
186+
* Determine if the connection is a cluster connection.
187+
*
188+
* @return bool
189+
*/
190+
public function isCluster()
191+
{
192+
return false;
193+
}
194+
185195
/**
186196
* Get the connection name.
187197
*
@@ -236,6 +246,25 @@ public function unsetEventDispatcher()
236246
$this->events = null;
237247
}
238248

249+
/**
250+
* Determine if the given key contains a Redis Cluster hash tag.
251+
*
252+
* @param string $key
253+
* @return bool
254+
*/
255+
public static function hasHashTag(string $key): bool
256+
{
257+
$open = strpos($key, '{');
258+
259+
if ($open === false) {
260+
return false;
261+
}
262+
263+
$close = strpos($key, '}', $open + 1);
264+
265+
return $close !== false && $close - $open > 1;
266+
}
267+
239268
/**
240269
* Pass other method calls down to the underlying client.
241270
*

‎src/Illuminate/Redis/Connections/PhpRedisClusterConnection.php‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,4 +78,15 @@ private function defaultNode()
7878

7979
return $this->defaultNode;
8080
}
81+
82+
/**
83+
* Determine if the connection is a cluster connection.
84+
*
85+
* @return bool
86+
*/
87+
#[\Override]
88+
public function isCluster()
89+
{
90+
return true;
91+
}
8192
}

‎src/Illuminate/Redis/Connections/PredisClusterConnection.php‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,4 +39,15 @@ public function flushdb()
3939
$node->executeCommand(tap(new $command)->setArguments(func_get_args()));
4040
}
4141
}
42+
43+
/**
44+
* Determine if the connection is a cluster connection.
45+
*
46+
* @return bool
47+
*/
48+
#[\Override]
49+
public function isCluster()
50+
{
51+
return true;
52+
}
4253
}

‎src/Illuminate/Redis/Connectors/PhpRedisConnector.php‎

Lines changed: 40 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -202,12 +202,14 @@ protected function createRedisClusterInstance(array $servers, array $options)
202202
isset($options['persistent']) && $options['persistent'],
203203
];
204204

205-
if (version_compare(phpversion('redis'), '4.3.0', '>=')) {
206-
$parameters[] = $options['password'] ?? null;
207-
}
205+
if (version_compare(phpversion('redis'), '5.3.2', '>=')) {
206+
$parameters[] = $this->formatClusterPassword($options);
208207

209-
if (version_compare(phpversion('redis'), '5.3.2', '>=') && ! is_null($context = Arr::get($options, 'context'))) {
210-
$parameters[] = $this->normalizeClusterContext($context);
208+
if (! is_null($context = Arr::get($options, 'context'))) {
209+
$parameters[] = $this->normalizeClusterContext($context);
210+
}
211+
} elseif (version_compare(phpversion('redis'), '4.3.0', '>=')) {
212+
$parameters[] = $options['password'] ?? null;
211213
}
212214

213215
return tap(new RedisCluster(...$parameters), function ($client) use ($options) {
@@ -238,6 +240,22 @@ protected function createRedisClusterInstance(array $servers, array $options)
238240
if (! empty($options['tcp_keepalive'])) {
239241
$client->setOption(Redis::OPT_TCP_KEEPALIVE, $options['tcp_keepalive']);
240242
}
243+
244+
if (array_key_exists('max_retries', $options)) {
245+
$client->setOption(Redis::OPT_MAX_RETRIES, $options['max_retries']);
246+
}
247+
248+
if (array_key_exists('backoff_algorithm', $options)) {
249+
$client->setOption(Redis::OPT_BACKOFF_ALGORITHM, $this->parseBackoffAlgorithm($options['backoff_algorithm']));
250+
}
251+
252+
if (array_key_exists('backoff_base', $options)) {
253+
$client->setOption(Redis::OPT_BACKOFF_BASE, $options['backoff_base']);
254+
}
255+
256+
if (array_key_exists('backoff_cap', $options)) {
257+
$client->setOption(Redis::OPT_BACKOFF_CAP, $options['backoff_cap']);
258+
}
241259
});
242260
}
243261

@@ -298,6 +316,23 @@ protected function normalizeClusterContext(array $context)
298316
return $context;
299317
}
300318

319+
/**
320+
* Format the password for a Redis cluster connection.
321+
*
322+
* @param array $options
323+
* @return string|array|null
324+
*/
325+
protected function formatClusterPassword(array $options)
326+
{
327+
$password = $options['password'] ?? null;
328+
329+
if (isset($options['username']) && $options['username'] !== '' && is_string($password)) {
330+
return [$options['username'], $password];
331+
}
332+
333+
return $password;
334+
}
335+
301336
/**
302337
* Parse a "friendly" backoff algorithm name into an integer.
303338
*

0 commit comments

Comments
 (0)