From c9494ab1245c71db3201ea53d1a7fde387e28c7c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=B3bert=20Kel=C4=8D=C3=A1k?= Date: Wed, 22 Jul 2026 11:26:19 +0200 Subject: [PATCH] Added stream commands to ClusterStrategy (#1708) * Added stream commands to ClusterStrategy * Fix edge case for XREADGROUP * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * Fix merge conflict --------- Co-authored-by: Vladyslav Vildanov <117659936+vladvildanov@users.noreply.github.com> Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- CHANGELOG.md | 1 + src/Cluster/ClusterStrategy.php | 77 ++++++++++++ tests/Predis/Cluster/PredisStrategyTest.php | 125 ++++++++++++++++++++ tests/Predis/Cluster/RedisStrategyTest.php | 122 +++++++++++++++++++ 4 files changed, 325 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 04db4139..c7bef6ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ - Added support for `TS.QUERYLABELS` command - Added support for `LMOVEM` and `BLMOVEM` commands - Added support for `FT.ALIASLIST` command +- Added stream commands to ClusterStrategy ### Fixed - Fixed Sentinel does not wipe servers on exception caused (#1694) diff --git a/src/Cluster/ClusterStrategy.php b/src/Cluster/ClusterStrategy.php index a7234d67..04f20c84 100644 --- a/src/Cluster/ClusterStrategy.php +++ b/src/Cluster/ClusterStrategy.php @@ -181,9 +181,25 @@ abstract class ClusterStrategy implements StrategyInterface 'HSTRLEN' => $getKeyFromFirstArgument, /* commands operating on streams */ + 'XACK' => $getKeyFromFirstArgument, + 'XACKDEL' => $getKeyFromFirstArgument, 'XADD' => $getKeyFromFirstArgument, + 'XAUTOCLAIM' => $getKeyFromFirstArgument, + 'XCFGSET' => $getKeyFromFirstArgument, + 'XCLAIM' => $getKeyFromFirstArgument, 'XDEL' => $getKeyFromFirstArgument, + 'XDELEX' => $getKeyFromFirstArgument, + 'XGROUP' => [$this, 'getKeyFromStreamGroupCommands'], + 'XINFO' => [$this, 'getKeyFromStreamGroupCommands'], + 'XLEN' => $getKeyFromFirstArgument, + 'XNACK' => $getKeyFromFirstArgument, + 'XPENDING' => $getKeyFromFirstArgument, 'XRANGE' => $getKeyFromFirstArgument, + 'XREAD' => [$this, 'getKeyFromStreamReadCommands'], + 'XREADGROUP' => [$this, 'getKeyFromStreamReadCommands'], + 'XREVRANGE' => $getKeyFromFirstArgument, + 'XSETID' => $getKeyFromFirstArgument, + 'XTRIM' => $getKeyFromFirstArgument, /* commands operating on time series */ 'TS.READ' => $getKeyFromFirstArgument, @@ -489,6 +505,67 @@ abstract class ClusterStrategy implements StrategyInterface return $this->getKeyFromAllArguments($command); } + /** + * Extracts the key from XGROUP and XINFO commands, where it follows the subcommand. + * + * @param CommandInterface $command Command instance. + * + * @return string|null + */ + protected function getKeyFromStreamGroupCommands(CommandInterface $command) + { + $arguments = $command->getArguments(); + + // Subcommands such as XINFO HELP and XGROUP HELP take no key at all. + if (!isset($arguments[1])) { + return null; + } + + return $arguments[1]; + } + + /** + * Extracts the key from XREAD and XREADGROUP commands, where the STREAMS token is followed by + * the same number of keys and IDs. + * + * @param CommandInterface $command Command instance. + * + * @return string|null + */ + protected function getKeyFromStreamReadCommands(CommandInterface $command) + { + $arguments = $command->getArguments(); + + $offset = $command->getId() === 'XREADGROUP' ? 3 : 0; + $position = null; + + for ($index = $offset, $argc = count($arguments); $index < $argc; ++$index) { + if (is_string($arguments[$index]) && strtoupper($arguments[$index]) === 'STREAMS') { + $position = $index; + break; + } + } + + if ($position === null) { + return null; + } + + $keysAndIds = array_slice($arguments, $position + 1); + $count = count($keysAndIds); + + if ($count === 0 || $count % 2 !== 0) { + return null; + } + + $keys = array_slice($keysAndIds, 0, intdiv($count, 2)); + + if (!$this->checkSameSlotForKeys($keys)) { + return null; + } + + return $keys[0]; + } + /** * Extracts the key from EVAL and EVALSHA commands. * diff --git a/tests/Predis/Cluster/PredisStrategyTest.php b/tests/Predis/Cluster/PredisStrategyTest.php index db7a2af8..92c82af4 100644 --- a/tests/Predis/Cluster/PredisStrategyTest.php +++ b/tests/Predis/Cluster/PredisStrategyTest.php @@ -222,6 +222,115 @@ class PredisStrategyTest extends PredisTestCase } } + /** + * @group disconnected + */ + public function testKeysForStreamCommands(): void + { + $strategy = $this->getClusterStrategy(); + $commands = $this->getCommandFactory(); + + // These commands drop or reshape their arguments unless all of them are given. + $arguments = [ + 'XACKDEL' => ['key', 'group', 'KEEPREF', ['1-1']], + 'XCLAIM' => ['key', 'group', 'consumer', 0, ['1-1']], + 'XDELEX' => ['key', 'KEEPREF', ['1-1']], + 'XNACK' => ['key', 'group', 'silent', ['1-1']], + 'XPENDING' => ['key', 'group'], + ]; + + foreach ($this->getExpectedCommands('keys-stream') as $commandID) { + $command = $commands->create($commandID, $arguments[$commandID]); + $this->assertNotNull($strategy->getSlot($command), $commandID); + } + } + + /** + * @group disconnected + */ + public function testKeysForStreamSubcommands(): void + { + $strategy = $this->getClusterStrategy(); + $commands = $this->getCommandFactory(); + + // The key of these commands follows the subcommand. + $arguments = [ + 'XGROUP' => ['CREATE', 'key', 'group', '$'], + 'XINFO' => ['GROUPS', 'key'], + ]; + + foreach ($this->getExpectedCommands('keys-stream-subcommand') as $commandID) { + $command = $commands->create($commandID, $arguments[$commandID]); + $this->assertNotNull($strategy->getSlot($command), $commandID); + } + } + + /** + * @group disconnected + */ + public function testReturnsNullOnStreamSubcommandsWithoutKey(): void + { + $strategy = $this->getClusterStrategy(); + $commands = $this->getCommandFactory(); + + foreach ($this->getExpectedCommands('keys-stream-subcommand') as $commandID) { + $command = $commands->create($commandID, ['HELP']); + $this->assertNull($strategy->getSlot($command), $commandID); + } + } + + /** + * @group disconnected + */ + public function testKeysForStreamReadCommands(): void + { + $strategy = $this->getClusterStrategy(); + $commands = $this->getCommandFactory(); + + // The keys follow the STREAMS token and are trailed by the same number of IDs. + $arguments = [ + 'XREAD' => [null, null, ['{key}:1', '{key}:2'], '0', '0'], + 'XREADGROUP' => ['group', 'consumer', null, null, false, '{key}:1', '{key}:2', '0', '0'], + ]; + + foreach ($this->getExpectedCommands('keys-stream-read') as $commandID) { + $command = $commands->create($commandID, $arguments[$commandID]); + $this->assertNotNull($strategy->getSlot($command), $commandID); + } + } + + /** + * @group disconnected + */ + public function testKeysForStreamReadCommandsWithReservedNames(): void + { + $strategy = $this->getClusterStrategy(); + $commands = $this->getCommandFactory(); + + // A group or a consumer may be named after the token that separates the keys. + $command = $commands->create('XREADGROUP', ['streams', 'streams', null, null, false, 'key', '0']); + $this->assertSame($strategy->getSlotByKey('key'), $strategy->getSlot($command)); + } + + /** + * @group disconnected + */ + public function testReturnsNullOnStreamReadCommandsWithDifferentSlots(): void + { + $strategy = $this->getClusterStrategy(); + $commands = $this->getCommandFactory(); + + $arguments = [ + 'XREAD' => [null, null, ['key:1', 'key:2'], '0', '0'], + 'XREADGROUP' => ['group', 'consumer', null, null, false, 'key:1', 'key:2', '0', '0'], + ]; + + foreach ($this->getExpectedCommands('keys-stream-read') as $commandID) { + $command = $commands->create($commandID, $arguments[$commandID]); + $this->assertNull($strategy->getSlot($command), $commandID); + } + } + /** * @group disconnected */ @@ -538,9 +647,25 @@ class PredisStrategyTest extends PredisTestCase 'HSTRLEN' => 'keys-first', /* commands operating on streams */ + 'XACK' => 'keys-first', + 'XACKDEL' => 'keys-stream', 'XADD' => 'keys-first', + 'XAUTOCLAIM' => 'keys-first', + 'XCFGSET' => 'keys-first', + 'XCLAIM' => 'keys-stream', 'XDEL' => 'keys-first', + 'XDELEX' => 'keys-stream', + 'XGROUP' => 'keys-stream-subcommand', + 'XINFO' => 'keys-stream-subcommand', + 'XLEN' => 'keys-first', + 'XNACK' => 'keys-stream', + 'XPENDING' => 'keys-stream', 'XRANGE' => 'keys-first', + 'XREAD' => 'keys-stream-read', + 'XREADGROUP' => 'keys-stream-read', + 'XREVRANGE' => 'keys-first', + 'XSETID' => 'keys-first', + 'XTRIM' => 'keys-first', /* commands operating on time series */ 'TS.READ' => 'keys-first', diff --git a/tests/Predis/Cluster/RedisStrategyTest.php b/tests/Predis/Cluster/RedisStrategyTest.php index 13013bd2..db7367a0 100644 --- a/tests/Predis/Cluster/RedisStrategyTest.php +++ b/tests/Predis/Cluster/RedisStrategyTest.php @@ -208,6 +208,80 @@ class RedisStrategyTest extends PredisTestCase /** * @group disconnected */ + public function testKeysForStreamCommands(): void + { + $strategy = $this->getClusterStrategy(); + $commands = $this->getCommandFactory(); + + // These commands drop or reshape their arguments unless all of them are given. + $arguments = [ + 'XACKDEL' => ['key', 'group', 'KEEPREF', ['1-1']], + 'XCLAIM' => ['key', 'group', 'consumer', 0, ['1-1']], + 'XDELEX' => ['key', 'KEEPREF', ['1-1']], + 'XNACK' => ['key', 'group', 'silent', ['1-1']], + 'XPENDING' => ['key', 'group'], + ]; + + foreach ($this->getExpectedCommands('keys-stream') as $commandID) { + $command = $commands->create($commandID, $arguments[$commandID]); + $this->assertNotNull($strategy->getSlot($command), $commandID); + } + } + + /** + * @group disconnected + */ + public function testKeysForStreamSubcommands(): void + { + $strategy = $this->getClusterStrategy(); + $commands = $this->getCommandFactory(); + + // The key of these commands follows the subcommand. + $arguments = [ + 'XGROUP' => ['CREATE', 'key', 'group', '$'], + 'XINFO' => ['GROUPS', 'key'], + ]; + + foreach ($this->getExpectedCommands('keys-stream-subcommand') as $commandID) { + $command = $commands->create($commandID, $arguments[$commandID]); + $this->assertNotNull($strategy->getSlot($command), $commandID); + } + } + + /** + * @group disconnected + */ + public function testReturnsNullOnStreamSubcommandsWithoutKey(): void + { + $strategy = $this->getClusterStrategy(); + $commands = $this->getCommandFactory(); + + foreach ($this->getExpectedCommands('keys-stream-subcommand') as $commandID) { + $command = $commands->create($commandID, ['HELP']); + $this->assertNull($strategy->getSlot($command), $commandID); + } + } + + /** + * @group disconnected + */ + public function testKeysForStreamReadCommands(): void + { + $strategy = $this->getClusterStrategy(); + $commands = $this->getCommandFactory(); + + // The keys follow the STREAMS token and are trailed by the same number of IDs. + $arguments = [ + 'XREAD' => [null, null, ['{key}:1', '{key}:2'], '0', '0'], + 'XREADGROUP' => ['group', 'consumer', null, null, false, '{key}:1', '{key}:2', '0', '0'], + ]; + + foreach ($this->getExpectedCommands('keys-stream-read') as $commandID) { + $command = $commands->create($commandID, $arguments[$commandID]); + $this->assertNotNull($strategy->getSlot($command), $commandID); + } + } + public function testKeysForFirstTwoKeysCommands(): void { $strategy = $this->getClusterStrategy(); @@ -220,6 +294,38 @@ class RedisStrategyTest extends PredisTestCase } } + /** + * @group disconnected + */ + public function testKeysForStreamReadCommandsWithReservedNames(): void + { + $strategy = $this->getClusterStrategy(); + $commands = $this->getCommandFactory(); + + // A group or a consumer may be named after the token that separates the keys. + $command = $commands->create('XREADGROUP', ['streams', 'streams', null, null, false, 'key', '0']); + $this->assertSame($strategy->getSlotByKey('key'), $strategy->getSlot($command)); + } + + /** + * @group disconnected + */ + public function testReturnsNullOnStreamReadCommandsWithDifferentSlots(): void + { + $strategy = $this->getClusterStrategy(); + $commands = $this->getCommandFactory(); + + $arguments = [ + 'XREAD' => [null, null, ['key:1', 'key:2'], '0', '0'], + 'XREADGROUP' => ['group', 'consumer', null, null, false, 'key:1', 'key:2', '0', '0'], + ]; + + foreach ($this->getExpectedCommands('keys-stream-read') as $commandID) { + $command = $commands->create($commandID, $arguments[$commandID]); + $this->assertNull($strategy->getSlot($command), $commandID); + } + } + /** * @group disconnected */ @@ -561,9 +667,25 @@ class RedisStrategyTest extends PredisTestCase 'HSTRLEN' => 'keys-first', /* commands operating on streams */ + 'XACK' => 'keys-first', + 'XACKDEL' => 'keys-stream', 'XADD' => 'keys-first', + 'XAUTOCLAIM' => 'keys-first', + 'XCFGSET' => 'keys-first', + 'XCLAIM' => 'keys-stream', 'XDEL' => 'keys-first', + 'XDELEX' => 'keys-stream', + 'XGROUP' => 'keys-stream-subcommand', + 'XINFO' => 'keys-stream-subcommand', + 'XLEN' => 'keys-first', + 'XNACK' => 'keys-stream', + 'XPENDING' => 'keys-stream', 'XRANGE' => 'keys-first', + 'XREAD' => 'keys-stream-read', + 'XREADGROUP' => 'keys-stream-read', + 'XREVRANGE' => 'keys-first', + 'XSETID' => 'keys-first', + 'XTRIM' => 'keys-first', /* commands operating on time series */ 'TS.READ' => 'keys-first',