Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Commit 7c416a7

Browse filesBrowse files
SeldaekSeldaek
committed
[RedisMessengerBridge] Add a delete_after_ack option to automatically clean up processed messages from memory
1 parent 6dc7d8b commit 7c416a7
Copy full SHA for 7c416a7

File tree

Expand file treeCollapse file tree

3 files changed

+42
-0
lines changed
Filter options
Expand file treeCollapse file tree

3 files changed

+42
-0
lines changed

‎src/Symfony/Component/Messenger/Bridge/Redis/CHANGELOG.md

Copy file name to clipboardExpand all lines: src/Symfony/Component/Messenger/Bridge/Redis/CHANGELOG.md
+2Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,3 +9,5 @@ CHANGELOG
99
* Deprecated use of invalid options
1010
* Added ability to receive of old pending messages with new `redeliver_timeout`
1111
and `claim_interval` options.
12+
* Added a `delete_after_ack` option to the DSN as an alternative to
13+
`stream_max_entries` to avoid leaking memory.

‎src/Symfony/Component/Messenger/Bridge/Redis/Tests/Transport/ConnectionTest.php

Copy file name to clipboardExpand all lines: src/Symfony/Component/Messenger/Bridge/Redis/Tests/Transport/ConnectionTest.php
+15Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -307,6 +307,21 @@ public function testMaxEntries()
307307
$connection->add('1', []);
308308
}
309309

310+
public function testDeleteAfterAck()
311+
{
312+
$redis = $this->getMockBuilder(\Redis::class)->disableOriginalConstructor()->getMock();
313+
314+
$redis->expects($this->exactly(1))->method('xack')
315+
->with('queue', 'symfony', ['1'])
316+
->willReturn(1);
317+
$redis->expects($this->exactly(1))->method('xdel')
318+
->with('queue', ['1'])
319+
->willReturn(1);
320+
321+
$connection = Connection::fromDsn('redis://localhost/queue?delete_after_ack=true', [], $redis); // 1 = always
322+
$connection->ack('1');
323+
}
324+
310325
public function testLastErrorGetsCleared()
311326
{
312327
$redis = $this->getMockBuilder(\Redis::class)->disableOriginalConstructor()->getMock();

‎src/Symfony/Component/Messenger/Bridge/Redis/Transport/Connection.php

Copy file name to clipboardExpand all lines: src/Symfony/Component/Messenger/Bridge/Redis/Transport/Connection.php
+25Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ class Connection
3232
'group' => 'symfony',
3333
'consumer' => 'consumer',
3434
'auto_setup' => true,
35+
'delete_after_ack' => false,
3536
'stream_max_entries' => 0, // any value higher than 0 defines an approximate maximum number of stream entries
3637
'dbindex' => 0,
3738
'tls' => false,
@@ -49,6 +50,7 @@ class Connection
4950
private $redeliverTimeout;
5051
private $nextClaim = 0;
5152
private $claimInterval;
53+
private $deleteAfterAck;
5254
private $couldHavePendingMessages = true;
5355

5456
public function __construct(array $configuration, array $connectionCredentials = [], array $redisOptions = [], \Redis $redis = null)
@@ -81,6 +83,7 @@ public function __construct(array $configuration, array $connectionCredentials =
8183
$this->queue = $this->stream.'__queue';
8284
$this->autoSetup = $configuration['auto_setup'] ?? self::DEFAULT_OPTIONS['auto_setup'];
8385
$this->maxEntries = $configuration['stream_max_entries'] ?? self::DEFAULT_OPTIONS['stream_max_entries'];
86+
$this->deleteAfterAck = $configuration['delete_after_ack'] ?? self::DEFAULT_OPTIONS['delete_after_ack'];
8487
$this->redeliverTimeout = ($configuration['redeliver_timeout'] ?? self::DEFAULT_OPTIONS['redeliver_timeout']) * 1000;
8588
$this->claimInterval = $configuration['claim_interval'] ?? self::DEFAULT_OPTIONS['claim_interval'];
8689
}
@@ -114,6 +117,12 @@ public static function fromDsn(string $dsn, array $redisOptions = [], \Redis $re
114117
unset($redisOptions['stream_max_entries']);
115118
}
116119

120+
$deleteAfterAck = null;
121+
if (\array_key_exists('delete_after_ack', $redisOptions)) {
122+
$deleteAfterAck = filter_var($redisOptions['delete_after_ack'], FILTER_VALIDATE_BOOLEAN);
123+
unset($redisOptions['delete_after_ack']);
124+
}
125+
117126
$dbIndex = null;
118127
if (\array_key_exists('dbindex', $redisOptions)) {
119128
$dbIndex = filter_var($redisOptions['dbindex'], FILTER_VALIDATE_INT);
@@ -144,6 +153,7 @@ public static function fromDsn(string $dsn, array $redisOptions = [], \Redis $re
144153
'consumer' => $redisOptions['consumer'] ?? null,
145154
'auto_setup' => $autoSetup,
146155
'stream_max_entries' => $maxEntries,
156+
'delete_after_ack' => $deleteAfterAck,
147157
'dbindex' => $dbIndex,
148158
'redeliver_timeout' => $redeliverTimeout,
149159
'claim_interval' => $claimInterval,
@@ -314,6 +324,9 @@ public function ack(string $id): void
314324
{
315325
try {
316326
$acknowledged = $this->connection->xack($this->stream, $this->group, [$id]);
327+
if ($this->deleteAfterAck) {
328+
$acknowledged = $this->connection->xdel($this->stream, [$id]);
329+
}
317330
} catch (\RedisException $e) {
318331
throw new TransportException($e->getMessage(), 0, $e);
319332
}
@@ -408,6 +421,18 @@ public function setup(): void
408421
$this->connection->clearLastError();
409422
}
410423

424+
if ($this->deleteAfterAck) {
425+
$groups = $this->connection->xinfo('GROUPS', $this->stream);
426+
if (
427+
// support for Redis extension version 5+
428+
(\is_array($groups) && 1 < \count($groups))
429+
// support for Redis extension version 4.x
430+
|| (\is_string($groups) && substr_count($groups, '"name"'))
431+
) {
432+
throw new LogicException(sprintf('More than one group exists for stream "%s", delete_after_ack can not be enabled as it risks deleting messages before all groups could consume them.', $this->stream));
433+
}
434+
}
435+
411436
$this->autoSetup = false;
412437
}
413438

0 commit comments

Comments
0 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.