geckon01/predis
1
0
Fork 0
mirror of https://github.com/predis/predis.git synced 2026-08-30 20:21:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Added temporary XREADGROUP_CLAIM command (#1608)

Browse Source 
* Added temporary XREADGROUP_CLAIM command

* Updated CHANGELOG.md

* Added deprecation annotation
This commit is contained in:
Vladyslav Vildanov
2025-11-12 09:14:57 +02:00
committed by GitHub
parent f11e855047
commit d0f37f0bac
10 changed files with 471 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+1
View File
@@ -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
src/ClientContextInterface.php
+1
View File
@@ -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)
src/ClientInterface.php
+1
View File
@@ -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)
src/Command/Redis/XGROUP.php
+8 -1
View File
@@ -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);
}
}
src/Command/Redis/XREADGROUP.php
+19 -1
View File
@@ -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);
}
}
src/Command/Redis/XREADGROUP_CLAIM.php
+81
View File
@@ -0,0 +1,81 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2025 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Command\Redis;
use Predis\Command\PrefixableCommand as RedisCommand;
/**
* This is a transitional command. In the next major version this command will replace XREADGROUP.
*/
class XREADGROUP_CLAIM extends RedisCommand
{
public function getId()
{
return 'XREADGROUP';
}
public function setArguments(array $arguments)
{
$processedArguments = ['GROUP', $arguments[0], $arguments[1]];
if (count($arguments) >= 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);
}
}
tests/PHPUnit/PredisTestCase.php
+8 -1
View File
@@ -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'] : []
);
tests/Predis/Command/Processor/KeyPrefixProcessorTest.php
+12
View File
@@ -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'],
tests/Predis/Command/Redis/XREADGROUP_CLAIM_Test.php
+291
View File
@@ -0,0 +1,291 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2025 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Command\Redis;
class XREADGROUP_CLAIM_Test extends PredisCommandTestCase
{
/**
* {@inheritDoc}
*/
protected function getExpectedCommand(): string
{
return XREADGROUP_CLAIM::class;
}
/**
* {@inheritDoc}
*/
protected function getExpectedId(): string
{
return 'XREADGROUP';
}
/**
* @dataProvider argumentsProvider
* @group disconnected
*/
public function testFilterArguments(array $actualArguments, array $expectedArguments): void
{
$command = $this->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'],
],
];
}
}
tests/Predis/Command/Redis/XREADGROUP_Test.php
+49
View File
@@ -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