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>
2026-09-04 23:08:00 +00:00 · 2026-07-22 11:26:19 +02:00
parent 422990b631
commit c9494ab124
4 changed files with 325 additions and 0 deletions
@@ -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)
@@ -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.
     *
@@ -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',
@@ -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',