[RedisMessengerBridge] Add a delete_after_ack option to automatically clean up processed messages from memory
This commit is contained in:
parent
6dc7d8b211
commit
7c416a7173
@ -9,3 +9,5 @@ CHANGELOG
|
||||
* Deprecated use of invalid options
|
||||
* Added ability to receive of old pending messages with new `redeliver_timeout`
|
||||
and `claim_interval` options.
|
||||
* Added a `delete_after_ack` option to the DSN as an alternative to
|
||||
`stream_max_entries` to avoid leaking memory.
|
||||
|
@ -307,6 +307,21 @@ class ConnectionTest extends TestCase
|
||||
$connection->add('1', []);
|
||||
}
|
||||
|
||||
public function testDeleteAfterAck()
|
||||
{
|
||||
$redis = $this->getMockBuilder(\Redis::class)->disableOriginalConstructor()->getMock();
|
||||
|
||||
$redis->expects($this->exactly(1))->method('xack')
|
||||
->with('queue', 'symfony', ['1'])
|
||||
->willReturn(1);
|
||||
$redis->expects($this->exactly(1))->method('xdel')
|
||||
->with('queue', ['1'])
|
||||
->willReturn(1);
|
||||
|
||||
$connection = Connection::fromDsn('redis://localhost/queue?delete_after_ack=true', [], $redis); // 1 = always
|
||||
$connection->ack('1');
|
||||
}
|
||||
|
||||
public function testLastErrorGetsCleared()
|
||||
{
|
||||
$redis = $this->getMockBuilder(\Redis::class)->disableOriginalConstructor()->getMock();
|
||||
|
@ -32,6 +32,7 @@ class Connection
|
||||
'group' => 'symfony',
|
||||
'consumer' => 'consumer',
|
||||
'auto_setup' => true,
|
||||
'delete_after_ack' => false,
|
||||
'stream_max_entries' => 0, // any value higher than 0 defines an approximate maximum number of stream entries
|
||||
'dbindex' => 0,
|
||||
'tls' => false,
|
||||
@ -49,6 +50,7 @@ class Connection
|
||||
private $redeliverTimeout;
|
||||
private $nextClaim = 0;
|
||||
private $claimInterval;
|
||||
private $deleteAfterAck;
|
||||
private $couldHavePendingMessages = true;
|
||||
|
||||
public function __construct(array $configuration, array $connectionCredentials = [], array $redisOptions = [], \Redis $redis = null)
|
||||
@ -81,6 +83,7 @@ class Connection
|
||||
$this->queue = $this->stream.'__queue';
|
||||
$this->autoSetup = $configuration['auto_setup'] ?? self::DEFAULT_OPTIONS['auto_setup'];
|
||||
$this->maxEntries = $configuration['stream_max_entries'] ?? self::DEFAULT_OPTIONS['stream_max_entries'];
|
||||
$this->deleteAfterAck = $configuration['delete_after_ack'] ?? self::DEFAULT_OPTIONS['delete_after_ack'];
|
||||
$this->redeliverTimeout = ($configuration['redeliver_timeout'] ?? self::DEFAULT_OPTIONS['redeliver_timeout']) * 1000;
|
||||
$this->claimInterval = $configuration['claim_interval'] ?? self::DEFAULT_OPTIONS['claim_interval'];
|
||||
}
|
||||
@ -114,6 +117,12 @@ class Connection
|
||||
unset($redisOptions['stream_max_entries']);
|
||||
}
|
||||
|
||||
$deleteAfterAck = null;
|
||||
if (\array_key_exists('delete_after_ack', $redisOptions)) {
|
||||
$deleteAfterAck = filter_var($redisOptions['delete_after_ack'], FILTER_VALIDATE_BOOLEAN);
|
||||
unset($redisOptions['delete_after_ack']);
|
||||
}
|
||||
|
||||
$dbIndex = null;
|
||||
if (\array_key_exists('dbindex', $redisOptions)) {
|
||||
$dbIndex = filter_var($redisOptions['dbindex'], FILTER_VALIDATE_INT);
|
||||
@ -144,6 +153,7 @@ class Connection
|
||||
'consumer' => $redisOptions['consumer'] ?? null,
|
||||
'auto_setup' => $autoSetup,
|
||||
'stream_max_entries' => $maxEntries,
|
||||
'delete_after_ack' => $deleteAfterAck,
|
||||
'dbindex' => $dbIndex,
|
||||
'redeliver_timeout' => $redeliverTimeout,
|
||||
'claim_interval' => $claimInterval,
|
||||
@ -314,6 +324,9 @@ class Connection
|
||||
{
|
||||
try {
|
||||
$acknowledged = $this->connection->xack($this->stream, $this->group, [$id]);
|
||||
if ($this->deleteAfterAck) {
|
||||
$acknowledged = $this->connection->xdel($this->stream, [$id]);
|
||||
}
|
||||
} catch (\RedisException $e) {
|
||||
throw new TransportException($e->getMessage(), 0, $e);
|
||||
}
|
||||
@ -408,6 +421,18 @@ class Connection
|
||||
$this->connection->clearLastError();
|
||||
}
|
||||
|
||||
if ($this->deleteAfterAck) {
|
||||
$groups = $this->connection->xinfo('GROUPS', $this->stream);
|
||||
if (
|
||||
// support for Redis extension version 5+
|
||||
(\is_array($groups) && 1 < \count($groups))
|
||||
// support for Redis extension version 4.x
|
||||
|| (\is_string($groups) && substr_count($groups, '"name"'))
|
||||
) {
|
||||
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));
|
||||
}
|
||||
}
|
||||
|
||||
$this->autoSetup = false;
|
||||
}
|
||||
|
||||
|
Reference in New Issue
Block a user