From 622f4df2b66ce4ce5557f672ce186c270e19f986 Mon Sep 17 00:00:00 2001 From: Vladyslav Vildanov <117659936+vladvildanov@users.noreply.github.com> Date: Wed, 22 Apr 2026 10:30:57 +0300 Subject: [PATCH] Added XNACK support (#1666) * Renamed GCRA parameters to match latest API * Added support for XNACK command * Updated test image * Updated test image * Updated version requirements * Fixed broken test * Updated CHANGELOG.md * Updated CHANGELOG.md --- CHANGELOG.md | 1 + src/ClientContextInterface.php | 1 + src/ClientInterface.php | 1 + src/Command/Redis/XNACK.php | 78 ++++++ tests/Predis/Command/Redis/XNACK_Test.php | 288 ++++++++++++++++++++++ 5 files changed, 369 insertions(+) create mode 100644 src/Command/Redis/XNACK.php create mode 100644 tests/Predis/Command/Redis/XNACK_Test.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 8fece12a..3af6cdb5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ - Added support for `GCRA` command (#1657) - Handle Redis Cluster `-READONLY` responses failover events (#1656) - Added FPHA argument for JSON.SET command (#1661) +- Added XNACK support (#1666) ### Changed - Include command name in unsupported container command error messages (#1653) diff --git a/src/ClientContextInterface.php b/src/ClientContextInterface.php index fe7ce8a2..95ba3758 100644 --- a/src/ClientContextInterface.php +++ b/src/ClientContextInterface.php @@ -301,6 +301,7 @@ use Predis\Command\Redis\VADD; * @method $this xdel(string $key, string ...$id) * @method $this xdelex(string $key, string $mode, array $ids) * @method $this xlen(string $key) + * @method $this xnack(string $key, string $group, string $mode, array $ids, ?int $retryCount = null, bool $force = false) * @method $this xpending(string $key, string $group, ?int $minIdleTime = null, ?string $start = null, ?string $end = null, ?int $count = null, ?string $consumer = null) * @method $this xrevrange(string $key, string $end, string $start, ?int $count = null) * @method $this xrange(string $key, string $start, string $end, ?int $count = null) diff --git a/src/ClientInterface.php b/src/ClientInterface.php index e04730bb..b8751b5d 100644 --- a/src/ClientInterface.php +++ b/src/ClientInterface.php @@ -312,6 +312,7 @@ use Predis\Response\Status; * @method int xdel(string $key, string ...$id) * @method array xdelex(string $key, string $mode, array $ids) * @method int xlen(string $key) + * @method int xnack(string $key, string $group, string $mode, array $ids, ?int $retryCount = null, bool $force = false) * @method array xpending(string $key, string $group, ?int $minIdleTime = null, ?string $start = null, ?string $end = null, ?int $count = null, ?string $consumer = null) * @method array xrevrange(string $key, string $end, string $start, ?int $count = null) * @method array xrange(string $key, string $start, string $end, ?int $count = null) diff --git a/src/Command/Redis/XNACK.php b/src/Command/Redis/XNACK.php new file mode 100644 index 00000000..6196d291 --- /dev/null +++ b/src/Command/Redis/XNACK.php @@ -0,0 +1,78 @@ + 'SILENT', + self::FAIL => 'FAIL', + self::FATAL => 'FATAL', + ]; + + /** + * {@inheritdoc} + */ + public function getId() + { + return 'XNACK'; + } + + /** + * {@inheritdoc} + */ + public function setArguments(array $arguments) + { + // arguments: [key, group, mode, ids, retryCount?, force?] + if (!in_array(strtoupper($arguments[2]), self::$modeEnum, true)) { + $enumValues = implode(', ', array_keys(self::$modeEnum)); + throw new UnexpectedValueException("Mode argument accepts only: {$enumValues} values"); + } + + $processedArguments = [ + $arguments[0], + $arguments[1], + strtoupper($arguments[2]), + ]; + + array_push($processedArguments, 'IDS', strval(count($arguments[3])), ...$arguments[3]); + + if (isset($arguments[4]) && $arguments[4] !== null) { + array_push($processedArguments, 'RETRYCOUNT', $arguments[4]); + } + + if (!empty($arguments[5])) { + $processedArguments[] = 'FORCE'; + } + + parent::setArguments($processedArguments); + } + + public function prefixKeys($prefix) + { + $this->applyPrefixForFirstArgument($prefix); + } +} diff --git a/tests/Predis/Command/Redis/XNACK_Test.php b/tests/Predis/Command/Redis/XNACK_Test.php new file mode 100644 index 00000000..f9159fb7 --- /dev/null +++ b/tests/Predis/Command/Redis/XNACK_Test.php @@ -0,0 +1,288 @@ +getCommand(); + $command->setArguments($actualArguments); + + $this->assertSame($expectedArguments, $command->getArguments()); + } + + /** + * @group disconnected + */ + public function testFilterArgumentsThrowsOnInvalidMode(): void + { + $this->expectException('UnexpectedValueException'); + $this->expectExceptionMessage('Mode argument accepts only: silent, fail, fatal values'); + + $command = $this->getCommand(); + $command->setArguments(['mystream', 'mygroup', 'INVALID', ['0-1']]); + } + + /** + * @group disconnected + */ + public function testParseResponse(): void + { + $this->assertSame(2, $this->getCommand()->parseResponse(2)); + } + + /** + * @group disconnected + */ + public function testPrefixKeys(): void + { + /** @var PrefixableCommand $command */ + $command = $this->getCommand(); + $actualArguments = ['mystream', 'mygroup', 'SILENT', ['1526569498055-0', '1526569498055-1']]; + $prefix = 'prefix:'; + $expectedArguments = ['prefix:mystream', 'mygroup', 'SILENT', 'IDS', '2', '1526569498055-0', '1526569498055-1']; + + $command->setArguments($actualArguments); + $command->prefixKeys($prefix); + + $this->assertSame($expectedArguments, $command->getArguments()); + } + + public function argumentsProvider(): array + { + return [ + 'with SILENT mode' => [ + ['mystream', 'mygroup', 'SILENT', ['1526569498055-0', '1526569498055-1']], + ['mystream', 'mygroup', 'SILENT', 'IDS', '2', '1526569498055-0', '1526569498055-1'], + ], + 'with FAIL mode' => [ + ['mystream', 'mygroup', 'FAIL', ['1526569498055-0']], + ['mystream', 'mygroup', 'FAIL', 'IDS', '1', '1526569498055-0'], + ], + 'with FATAL mode' => [ + ['mystream', 'mygroup', 'FATAL', ['1526569498055-0']], + ['mystream', 'mygroup', 'FATAL', 'IDS', '1', '1526569498055-0'], + ], + 'with lowercase mode normalized to uppercase' => [ + ['mystream', 'mygroup', 'silent', ['1526569498055-0']], + ['mystream', 'mygroup', 'SILENT', 'IDS', '1', '1526569498055-0'], + ], + 'with RETRYCOUNT option' => [ + ['mystream', 'mygroup', 'FAIL', ['1526569498055-0'], 3], + ['mystream', 'mygroup', 'FAIL', 'IDS', '1', '1526569498055-0', 'RETRYCOUNT', 3], + ], + 'with RETRYCOUNT zero' => [ + ['mystream', 'mygroup', 'SILENT', ['1526569498055-0'], 0], + ['mystream', 'mygroup', 'SILENT', 'IDS', '1', '1526569498055-0', 'RETRYCOUNT', 0], + ], + 'with FORCE flag' => [ + ['mystream', 'mygroup', 'FAIL', ['1526569498055-0'], null, true], + ['mystream', 'mygroup', 'FAIL', 'IDS', '1', '1526569498055-0', 'FORCE'], + ], + 'with RETRYCOUNT and FORCE' => [ + ['mystream', 'mygroup', 'FATAL', ['1526569498055-0', '1526569498055-1'], 5, true], + ['mystream', 'mygroup', 'FATAL', 'IDS', '2', '1526569498055-0', '1526569498055-1', 'RETRYCOUNT', 5, 'FORCE'], + ], + ]; + } + + /** + * @group connected + * @requiresRedisVersion >= 8.7.2 + */ + public function testNackWithSilentMode(): void + { + $redis = $this->getClient(); + + $redis->xadd('stream', ['key0' => 'val0'], '0-1'); + $redis->xadd('stream', ['key1' => 'val1'], '1-1'); + $redis->xgroup->create('stream', 'grp', '0'); + $redis->xreadgroup('grp', 'consumer1', 2, null, false, 'stream', '>'); + + $result = $redis->xnack('stream', 'grp', 'SILENT', ['0-1', '1-1']); + + $this->assertSame(2, $result); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.7.2 + */ + public function testNackWithFailMode(): void + { + $redis = $this->getClient(); + + $redis->xadd('stream', ['key0' => 'val0'], '0-1'); + $redis->xgroup->create('stream', 'grp', '0'); + $redis->xreadgroup('grp', 'consumer1', 1, null, false, 'stream', '>'); + + $result = $redis->xnack('stream', 'grp', 'FAIL', ['0-1']); + + $this->assertSame(1, $result); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.7.2 + */ + public function testNackWithFatalMode(): void + { + $redis = $this->getClient(); + + $redis->xadd('stream', ['key0' => 'val0'], '0-1'); + $redis->xgroup->create('stream', 'grp', '0'); + $redis->xreadgroup('grp', 'consumer1', 1, null, false, 'stream', '>'); + + $result = $redis->xnack('stream', 'grp', 'FATAL', ['0-1']); + + $this->assertSame(1, $result); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.7.2 + */ + public function testNackWithSomeIdsNotInPel(): void + { + $redis = $this->getClient(); + + $redis->xadd('stream', ['key0' => 'val0'], '0-1'); + $redis->xadd('stream', ['key1' => 'val1'], '1-1'); + $redis->xgroup->create('stream', 'grp', '0'); + $redis->xreadgroup('grp', 'consumer1', 1, null, false, 'stream', '>'); + + // Only 0-1 is in PEL, 1-1 is not claimed yet + $result = $redis->xnack('stream', 'grp', 'FAIL', ['0-1', '1-1', '9-9']); + + $this->assertSame(1, $result); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.7.2 + */ + public function testNackWithRetryCount(): void + { + $redis = $this->getClient(); + + $redis->xadd('stream', ['key0' => 'val0'], '0-1'); + $redis->xgroup->create('stream', 'grp', '0'); + $redis->xreadgroup('grp', 'consumer1', 1, null, false, 'stream', '>'); + + $result = $redis->xnack('stream', 'grp', 'FAIL', ['0-1'], 5); + + $this->assertSame(1, $result); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.7.2 + */ + public function testNackWithForceFlag(): void + { + $redis = $this->getClient(); + + $redis->xadd('stream', ['key0' => 'val0'], '0-1'); + $redis->xgroup->create('stream', 'grp', '0'); + + // FORCE creates PEL entry even if not claimed + $result = $redis->xnack('stream', 'grp', 'FAIL', ['0-1'], null, true); + + $this->assertSame(1, $result); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.7.2 + */ + public function testNackOnNonExistentStreamReturnsError(): void + { + $this->expectException('Predis\Response\ServerException'); + $this->expectExceptionMessageMatches('/NOGROUP/'); + + $redis = $this->getClient(); + $redis->xnack('nonexistent', 'grp', 'FAIL', ['0-1']); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.7.2 + */ + public function testNackWithNonExistentGroupReturnsError(): void + { + $this->expectException('Predis\Response\ServerException'); + $this->expectExceptionMessageMatches('/NOGROUP/'); + + $redis = $this->getClient(); + $redis->xadd('stream', ['key0' => 'val0'], '0-1'); + $redis->xnack('stream', 'nonexistent', 'FAIL', ['0-1']); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.7.2 + */ + public function testNackOnWrongTypeReturnsError(): void + { + $this->expectException('Predis\Response\ServerException'); + $this->expectExceptionMessageMatches('/WRONGTYPE/'); + + $redis = $this->getClient(); + $redis->set('notastream', 'somevalue'); + $redis->xnack('notastream', 'grp', 'FAIL', ['0-1']); + } + + /** + * @group connected + * @requiresRedisVersion >= 8.7.2 + */ + public function testNackWithResp3Protocol(): void + { + $redis = $this->getResp3Client(); + + $redis->xadd('stream', ['key0' => 'val0'], '0-1'); + $redis->xgroup->create('stream', 'grp', '0'); + $redis->xreadgroup('grp', 'consumer1', 1, null, false, 'stream', '>'); + + $result = $redis->xnack('stream', 'grp', 'SILENT', ['0-1']); + + $this->assertSame(1, $result); + } +}