From d0f37f0bac1a22b2faf01a8e92d9227fc18d8243 Mon Sep 17 00:00:00 2001 From: Vladyslav Vildanov <117659936+vladvildanov@users.noreply.github.com> Date: Wed, 12 Nov 2025 09:14:57 +0200 Subject: [PATCH] Added temporary XREADGROUP_CLAIM command (#1608) * Added temporary XREADGROUP_CLAIM command * Updated CHANGELOG.md * Added deprecation annotation --- CHANGELOG.md | 1 + src/ClientContextInterface.php | 1 + src/ClientInterface.php | 1 + src/Command/Redis/XGROUP.php | 9 +- src/Command/Redis/XREADGROUP.php | 20 +- src/Command/Redis/XREADGROUP_CLAIM.php | 81 +++++ tests/PHPUnit/PredisTestCase.php | 9 +- .../Processor/KeyPrefixProcessorTest.php | 12 + .../Command/Redis/XREADGROUP_CLAIM_Test.php | 291 ++++++++++++++++++ .../Predis/Command/Redis/XREADGROUP_Test.php | 49 +++ 10 files changed, 471 insertions(+), 3 deletions(-) create mode 100644 src/Command/Redis/XREADGROUP_CLAIM.php create mode 100644 tests/Predis/Command/Redis/XREADGROUP_CLAIM_Test.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 00b5bd4b..8ac9c427 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ ### Added - Added cluster support for `XADD`, `XDEL` and `XRANGE` (#1587) - Added prefixable interface for `HEXPIRE` and `HEXPIRETIME` (#1592) +- Added temporary XREADGROUP_CLAIM command (#1608) - Added support for MSET command (#1610) ### Changed diff --git a/src/ClientContextInterface.php b/src/ClientContextInterface.php index a215e83a..f7b512f9 100644 --- a/src/ClientContextInterface.php +++ b/src/ClientContextInterface.php @@ -300,6 +300,7 @@ use Predis\Command\Redis\VADD; * @method $this xrange(string $key, string $start, string $end, ?int $count = null) * @method $this xread(int $count = null, int $block = null, array $streams = null, string ...$id) * @method $this xreadgroup(string $group, string $consumer, ?int $count = null, ?int $blockMs = null, bool $noAck = false, string ...$keyOrId) + * @method $this xreadgroup_claim(string $group, string $consumer, array $keyIdDict, ?int $count = null, ?int $blockMs = null, bool $noAck = false, ?int $claim = null) * @method $this xsetid(string $key, string $lastId, ?int $entriesAdded = null, ?string $maxDeleteId = null) * @method $this xtrim(string $key, array|string $strategy, string $threshold, array $options = null) * @method $this zadd($key, array $membersAndScoresDictionary) diff --git a/src/ClientInterface.php b/src/ClientInterface.php index 08c24d51..c8bc7b8f 100644 --- a/src/ClientInterface.php +++ b/src/ClientInterface.php @@ -311,6 +311,7 @@ use Predis\Response\Status; * @method array xrange(string $key, string $start, string $end, ?int $count = null) * @method array|null xread(int $count = null, int $block = null, array $streams = null, string ...$id) * @method array xreadgroup(string $group, string $consumer, ?int $count = null, ?int $blockMs = null, bool $noAck = false, string ...$keyOrId) + * @method array xreadgroup_claim(string $group, string $consumer, array $keyIdDict, ?int $count = null, ?int $blockMs = null, bool $noAck = false, ?int $claim = null) * @method Status xsetid(string $key, string $lastId, ?int $entriesAdded = null, ?string $maxDeleteId = null) * @method string xtrim(string $key, array|string $strategy, string $threshold, array $options = null) * @method int zadd(string $key, array $membersAndScoresDictionary) diff --git a/src/Command/Redis/XGROUP.php b/src/Command/Redis/XGROUP.php index 79a3d0c3..7adc431f 100644 --- a/src/Command/Redis/XGROUP.php +++ b/src/Command/Redis/XGROUP.php @@ -12,7 +12,7 @@ namespace Predis\Command\Redis; -use Predis\Command\Command as RedisCommand; +use Predis\Command\PrefixableCommand as RedisCommand; /** * @see https://redis.io/commands/?name=xgroup @@ -78,4 +78,11 @@ class XGROUP extends RedisCommand parent::setArguments($processedArguments); } + + public function prefixKeys($prefix) + { + $arguments = $this->getArguments(); + $arguments[1] = $prefix . $arguments[1]; + $this->setRawArguments($arguments); + } } diff --git a/src/Command/Redis/XREADGROUP.php b/src/Command/Redis/XREADGROUP.php index 4fc85d2f..43d6264d 100644 --- a/src/Command/Redis/XREADGROUP.php +++ b/src/Command/Redis/XREADGROUP.php @@ -12,8 +12,12 @@ namespace Predis\Command\Redis; -use Predis\Command\Command as RedisCommand; +use Predis\Command\PrefixableCommand as RedisCommand; +/** + * @deprecated Public API will be changed in the next major version. + * XREADGROUP_CLAIM API will be used instead. + */ class XREADGROUP extends RedisCommand { public function getId() @@ -58,4 +62,18 @@ class XREADGROUP extends RedisCommand return $result; } + + public function prefixKeys($prefix) + { + $arguments = $this->getArguments(); + $keyIdsStartingIndex = array_search('STREAMS', $arguments) + 1; + $keysAndIdsCount = count($arguments) - $keyIdsStartingIndex; + $keysCount = $keysAndIdsCount / 2; + + for ($i = $keyIdsStartingIndex; $i < $keyIdsStartingIndex + $keysCount; $i++) { + $arguments[$i] = $prefix . $arguments[$i]; + } + + parent::setRawArguments($arguments); + } } diff --git a/src/Command/Redis/XREADGROUP_CLAIM.php b/src/Command/Redis/XREADGROUP_CLAIM.php new file mode 100644 index 00000000..c15f050d --- /dev/null +++ b/src/Command/Redis/XREADGROUP_CLAIM.php @@ -0,0 +1,81 @@ += 4 && null !== $arguments[3]) { + array_push($processedArguments, 'COUNT', $arguments[3]); + } + + if (count($arguments) >= 5 && null !== $arguments[4]) { + array_push($processedArguments, 'BLOCK', $arguments[4]); + } + + if (count($arguments) >= 6 && false !== $arguments[5]) { + $processedArguments[] = 'NOACK'; + } + + if (count($arguments) >= 7 && false !== $arguments[6]) { + array_push($processedArguments, 'CLAIM', $arguments[6]); + } + + array_push($processedArguments, 'STREAMS', ...array_keys($arguments[2]), ...array_values($arguments[2])); + + parent::setArguments($processedArguments); + } + + public function parseResponse($data) + { + if (!is_array($data) || $data === array_values($data)) { + return $data; + } + + // Relay + $result = []; + foreach ($data as $key => $value) { + $group = [$key, $value]; + $result[] = $group; + } + + return $result; + } + + public function prefixKeys($prefix) + { + $arguments = $this->getArguments(); + $keyIdsStartingIndex = array_search('STREAMS', $arguments) + 1; + $keysAndIdsCount = count($arguments) - $keyIdsStartingIndex; + $keysCount = $keysAndIdsCount / 2; + + for ($i = $keyIdsStartingIndex; $i < $keyIdsStartingIndex + $keysCount; $i++) { + $arguments[$i] = $prefix . $arguments[$i]; + } + + parent::setRawArguments($arguments); + } +} diff --git a/tests/PHPUnit/PredisTestCase.php b/tests/PHPUnit/PredisTestCase.php index 51a70746..77115211 100644 --- a/tests/PHPUnit/PredisTestCase.php +++ b/tests/PHPUnit/PredisTestCase.php @@ -16,6 +16,7 @@ use PHPUnit\OneOfConstraint; use PHPUnit\Util\Test as TestUtil; use Predis\Client; use Predis\Command; +use Predis\Command\Processor\KeyPrefixProcessor; use Predis\Connection; /** @@ -262,8 +263,14 @@ abstract class PredisTestCase extends PHPUnit\Framework\TestCase $parameters ?: [] ); + $commandsFactory = $this->getCommandFactory(); + + if (null !== $options && array_key_exists('prefix', $options)) { + $commandsFactory->setProcessor(new KeyPrefixProcessor($options['prefix'])); + } + $options = array_merge( - ['commands' => $this->getCommandFactory()], + ['commands' => $commandsFactory], $options ?: [], getenv('USE_RELAY') ? ['connections' => 'relay'] : [] ); diff --git a/tests/Predis/Command/Processor/KeyPrefixProcessorTest.php b/tests/Predis/Command/Processor/KeyPrefixProcessorTest.php index c6d1a627..c52a32c8 100644 --- a/tests/Predis/Command/Processor/KeyPrefixProcessorTest.php +++ b/tests/Predis/Command/Processor/KeyPrefixProcessorTest.php @@ -808,6 +808,18 @@ class KeyPrefixProcessorTest extends PredisTestCase ['key', 'MAXLEN', 100], ['prefix:key', 'MAXLEN', 100], ], + ['XGROUP', + ['CREATE', 'key', 'group', '$'], + ['CREATE', 'prefix:key', 'group', '$'], + ], + ['XREADGROUP', + ['group', 'consumer', 10, 10, true, 'stream', 'stream1', '0-0', '0-0'], + ['GROUP', 'group', 'consumer', 'COUNT', 10, 'BLOCK', 10, 'NOACK', 'STREAMS', 'prefix:stream', 'prefix:stream1', '0-0', '0-0'], + ], + ['XREADGROUP_CLAIM', + ['group', 'consumer', ['stream' => '0-0', 'stream1' => '0-0'], 10, 10, true, 10], + ['GROUP', 'group', 'consumer', 'COUNT', 10, 'BLOCK', 10, 'NOACK', 'CLAIM', 10, 'STREAMS', 'prefix:stream', 'prefix:stream1', '0-0', '0-0'], + ], ['ZPOPMIN', ['key'], ['prefix:key'], diff --git a/tests/Predis/Command/Redis/XREADGROUP_CLAIM_Test.php b/tests/Predis/Command/Redis/XREADGROUP_CLAIM_Test.php new file mode 100644 index 00000000..6c6cb295 --- /dev/null +++ b/tests/Predis/Command/Redis/XREADGROUP_CLAIM_Test.php @@ -0,0 +1,291 @@ +getCommand(); + $command->setArguments($actualArguments); + + $this->assertSame($expectedArguments, $command->getArguments()); + } + + /** + * @group connected + * @return void + * @requiresRedisVersion >= 5.0.0 + */ + public function testReadsFromGivenConsumerGroup(): void + { + $redis = $this->getClient(); + + $streamInitId = $redis->xadd('stream', ['field' => 'value']); + $this->assertEquals('OK', $redis->xgroup->create('stream', 'group', $streamInitId)); + + $nextId = $redis->xadd('stream', ['newField' => 'newValue']); + $expectedResponse = [ + [ + 'stream', + [ + [$nextId, ['newField', 'newValue']], + ], + ], + ]; + + $this->assertSame( + $expectedResponse, + $redis->xreadgroup_claim( + 'group', + 'consumer', + ['stream' => '>'] + ) + ); + } + + /** + * @group connected + * @return void + * @requiresRedisVersion >= 5.0.0 + */ + public function testReadsFromConsumerGroupFromMultipleStreams(): void + { + $redis = $this->getClient(); + + $streamInitId = $redis->xadd('stream', ['field' => 'value']); + $this->assertEquals('OK', $redis->xgroup->create('stream', 'group', $streamInitId)); + + $anotherStreamInitId = $redis->xadd('another_stream', ['field' => 'value']); + $this->assertEquals('OK', $redis->xgroup->create('another_stream', 'group', $anotherStreamInitId)); + + $nextId = $redis->xadd('stream', ['newField' => 'newValue']); + $anotherNextId = $redis->xadd('another_stream', ['newField' => 'newValue']); + + $expectedResponse = [ + [ + 'stream', + [ + [$nextId, ['newField', 'newValue']], + ], + ], + [ + 'another_stream', + [ + [$anotherNextId, ['newField', 'newValue']], + ], + ], + ]; + + $this->assertSame( + $expectedResponse, + $redis->xreadgroup_claim( + 'group', + 'consumer', + ['stream' => '>', 'another_stream' => '>'] + ) + ); + } + + /** + * @group connected + * @return void + * @requiresRedisVersion >= 5.0.0 + */ + public function testPrefixStreamKeys(): void + { + $redis = $this->createClient(null, ['prefix' => 'prefix:']); + + $streamInitId = $redis->xadd('stream', ['field' => 'value']); + $this->assertEquals('OK', $redis->xgroup->create('stream', 'group', $streamInitId)); + + $anotherStreamInitId = $redis->xadd('another_stream', ['field' => 'value']); + $this->assertEquals('OK', $redis->xgroup->create('another_stream', 'group', $anotherStreamInitId)); + + $nextId = $redis->xadd('stream', ['newField' => 'newValue']); + $anotherNextId = $redis->xadd('another_stream', ['newField' => 'newValue']); + + $expectedResponse = [ + [ + 'prefix:stream', + [ + [$nextId, ['newField', 'newValue']], + ], + ], + [ + 'prefix:another_stream', + [ + [$anotherNextId, ['newField', 'newValue']], + ], + ], + ]; + + $this->assertSame( + $expectedResponse, + $redis->xreadgroup_claim( + 'group', + 'consumer', + ['stream' => '>', 'another_stream' => '>'] + ) + ); + } + + /** + * @group connected + * @return void + * @requiresRedisVersion >= 8.3.224 + */ + public function testReadsAndClaimFromConsumerGroupFromSingleStream(): void + { + $redis = $this->getClient(); + + $streamInitId = $redis->xadd('stream', ['field' => 'value']); + $this->assertEquals('OK', $redis->xgroup->create('stream', 'group', $streamInitId)); + + $nextId = $redis->xadd('stream', ['newField' => 'newValue']); + + $expectedResponse = [ + [ + 'stream', + [ + [$nextId, ['newField', 'newValue'], '0', '0'], + ], + ], + ]; + + $this->assertEmpty( + $redis->xpending('stream', 'group', null, '-', '+', 5) + ); + + $this->assertEquals( + $expectedResponse, + $redis->xreadgroup_claim( + 'group', + 'consumer', + ['stream' => '>'], null, null, false, 10) + ); + + $this->assertCount( + 4, + $redis->xpending('stream', 'group', null, '-', '+', 5)[0] + ); + } + + /** + * @group connected + * @return void + * @requiresRedisVersion >= 8.3.224 + */ + public function testReadsAndClaimFromConsumerGroupFromMultipleStreams(): void + { + $redis = $this->getClient(); + + $streamInitId = $redis->xadd('stream', ['field' => 'value']); + $this->assertEquals('OK', $redis->xgroup->create('stream', 'group', $streamInitId)); + + $anotherStreamInitId = $redis->xadd('another_stream', ['field' => 'value']); + $this->assertEquals('OK', $redis->xgroup->create('another_stream', 'group', $anotherStreamInitId)); + + $nextId = $redis->xadd('stream', ['newField' => 'newValue']); + $anotherNextId = $redis->xadd('another_stream', ['newField' => 'newValue']); + + $expectedResponse = [ + [ + 'stream', + [ + [$nextId, ['newField', 'newValue'], '0', '0'], + ], + ], + [ + 'another_stream', + [ + [$anotherNextId, ['newField', 'newValue'], '0', '0'], + ], + ], + ]; + + $this->assertEmpty( + $redis->xpending('stream', 'group', null, '-', '+', 5) + ); + $this->assertEmpty( + $redis->xpending('another_stream', 'group', null, '-', '+', 5) + ); + + $this->assertEquals( + $expectedResponse, + $redis->xreadgroup_claim( + 'group', + 'consumer', + ['stream' => '>', 'another_stream' => '>'], null, null, false, 10) + ); + + $this->assertCount( + 4, + $redis->xpending('stream', 'group', null, '-', '+', 5)[0] + ); + $this->assertCount( + 4, + $redis->xpending('another_stream', 'group', null, '-', '+', 5)[0] + ); + } + + public function argumentsProvider(): array + { + return [ + 'with default arguments' => [ + ['group', 'consumer', ['stream' => '0-0']], + ['GROUP', 'group', 'consumer', 'STREAMS', 'stream', '0-0'], + ], + 'with COUNT modifier' => [ + ['group', 'consumer', ['stream' => '0-0'], 10], + ['GROUP', 'group', 'consumer', 'COUNT', 10, 'STREAMS', 'stream', '0-0'], + ], + 'with BLOCK modifier' => [ + ['group', 'consumer', ['stream' => '0-0'], null, 10], + ['GROUP', 'group', 'consumer', 'BLOCK', 10, 'STREAMS', 'stream', '0-0'], + ], + 'with NOACK modifier' => [ + ['group', 'consumer', ['stream' => '0-0'], null, null, true], + ['GROUP', 'group', 'consumer', 'NOACK', 'STREAMS', 'stream', '0-0'], + ], + 'with CLAIM modifier' => [ + ['group', 'consumer', ['stream' => '0-0'], null, null, false, 10], + ['GROUP', 'group', 'consumer', 'CLAIM', 10, 'STREAMS', 'stream', '0-0'], + ], + 'with all arguments' => [ + ['group', 'consumer', ['stream' => '0-0', 'stream1' => '0-0'], 10, 10, true, 20], + ['GROUP', 'group', 'consumer', 'COUNT', 10, 'BLOCK', 10, 'NOACK', 'CLAIM', 20, 'STREAMS', 'stream', 'stream1', '0-0', '0-0'], + ], + ]; + } +} diff --git a/tests/Predis/Command/Redis/XREADGROUP_Test.php b/tests/Predis/Command/Redis/XREADGROUP_Test.php index f796197f..7d0c0e99 100644 --- a/tests/Predis/Command/Redis/XREADGROUP_Test.php +++ b/tests/Predis/Command/Redis/XREADGROUP_Test.php @@ -136,6 +136,55 @@ class XREADGROUP_Test extends PredisCommandTestCase ); } + /** + * @group connected + * @return void + * @requiresRedisVersion >= 5.0.0 + */ + public function testPrefixStreamKeys(): void + { + $redis = $this->createClient(null, ['prefix' => 'prefix:']); + + $streamInitId = $redis->xadd('stream', ['field' => 'value']); + $this->assertEquals('OK', $redis->xgroup->create('stream', 'group', $streamInitId)); + + $anotherStreamInitId = $redis->xadd('another_stream', ['field' => 'value']); + $this->assertEquals('OK', $redis->xgroup->create('another_stream', 'group', $anotherStreamInitId)); + + $nextId = $redis->xadd('stream', ['newField' => 'newValue']); + $anotherNextId = $redis->xadd('another_stream', ['newField' => 'newValue']); + + $expectedResponse = [ + [ + 'prefix:stream', + [ + [$nextId, ['newField', 'newValue']], + ], + ], + [ + 'prefix:another_stream', + [ + [$anotherNextId, ['newField', 'newValue']], + ], + ], + ]; + + $this->assertSame( + $expectedResponse, + $redis->xreadgroup( + 'group', + 'consumer', + null, + null, + false, + 'stream', + 'another_stream', + '>', + '>' + ) + ); + } + /** * @group connected * @return void