implement XCLAIM command (#1557)

* implement XCLAIM command

* modify CHANGELOG.md

* add empty line before return

* test xclaim with all available options

* review refactoring

* remove fast-track check for empty answer

---------

Co-authored-by: aleksanders <alexander.s@seranking.com>

CHANGELOG.md

@@ -4,6 +4,7 @@
 ### Added
 - Add experimental support for vector sets commands (#1550)
 - Added support for `XACK` command (#1555)
+- Added support for `XCLAIM` command (#1557)
 
 ### Changed
 - Handle and retry `LOADING` errors from Sentinel replicas (#1536)

src/ClientContextInterface.php

@@ -285,6 +285,7 @@ use Predis\Command\Redis\VADD;
  * @method $this tsrange(string $key, $fromTimestamp, $toTimestamp, ?RangeArguments $arguments = null)
  * @method $this tsrevrange(string $key, $fromTimestamp, $toTimestamp, ?RangeArguments $arguments = null)
  * @method $this xack(string $key, string $group, string ...$id)
+ * @method $this xclaim(string $key, string $group, string $consumer, int $minIdleTime, string|array $ids, ?int $idle = null, ?int $time = null, ?int $retryCount = null, bool $force = false, bool $justId = false, ?string $lastId = null)
  * @method $this zadd($key, array $membersAndScoresDictionary)
  * @method $this zcard($key)
  * @method $this zcount($key, $min, $max)

src/ClientInterface.php

@@ -298,6 +298,7 @@ use Predis\Response\Status;
  * @method int               xack(string $key, string $group, string ...$id)
  * @method string            xadd(string $key, array $dictionary, string $id = '*', array $options = null)
  * @method array             xautoclaim(string $key, string $group, string $consumer, int $minIdleTime, string $start, ?int $count = null, bool $justId = false)
+ * @method array             xclaim(string $key, string $group, string $consumer, int $minIdleTime, string|array $ids, ?int $idle = null, ?int $time = null, ?int $retryCount = null, bool $force = false, bool $justId = false, ?string $lastId = null)
  * @method int               xdel(string $key, string ...$id)
  * @method int               xlen(string $key)
  * @method array             xrevrange(string $key, string $end, string $start, ?int $count = null)

src/Command/Redis/XCLAIM.php

@@ -0,0 +1,89 @@
+<?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;
+use Predis\Command\Redis\Utils\CommandUtility;
+
+/**
+ * @see http://redis.io/commands/xclaim
+ */
+class XCLAIM extends RedisCommand
+{
+    public function getId(): string
+    {
+        return 'XCLAIM';
+    }
+
+    public function setArguments(array $arguments): void
+    {
+        if (count($arguments) < 5) {
+            return;
+        }
+
+        $processedArguments = array_slice($arguments, 0, 4);
+        $ids = $arguments[4];
+        $processedArguments = array_merge($processedArguments, is_array($ids) ? $ids : [$ids]);
+
+        if (array_key_exists(5, $arguments) && null !== $arguments[5]) {
+            array_push($processedArguments, 'IDLE', $arguments[5]);
+        }
+
+        if (array_key_exists(6, $arguments) && null !== $arguments[6]) {
+            array_push($processedArguments, 'TIME', $arguments[6]);
+        }
+
+        if (array_key_exists(7, $arguments) && null !== $arguments[7]) {
+            array_push($processedArguments, 'RETRYCOUNT', $arguments[7]);
+        }
+
+        if (array_key_exists(8, $arguments) && false !== $arguments[8]) {
+            $processedArguments[] = 'FORCE';
+        }
+
+        if (array_key_exists(9, $arguments) && false !== $arguments[9]) {
+            $processedArguments[] = 'JUSTID';
+        }
+
+        if (array_key_exists(10, $arguments) && false !== $arguments[10]) {
+            array_push($processedArguments, 'LASTID', $arguments[10]);
+        }
+
+        parent::setArguments($processedArguments);
+    }
+
+    public function parseResponse($data): array
+    {
+        // JUSTID format
+        if (isset($data[0]) && !is_array($data[0])) {
+            return $data;
+        }
+
+        $result = [];
+        foreach ($data as [$id, $kvDict]) {
+            $result[$id] = CommandUtility::arrayToDictionary($kvDict);
+        }
+
+        return $result;
+    }
+
+    public function parseResp3Response($data): array
+    {
+        return $this->parseResponse($data);
+    }
+
+    public function prefixKeys($prefix)
+    {
+        $this->applyPrefixForFirstArgument($prefix);
+    }
+}

tests/Predis/Command/Redis/XCLAIM_Test.php

@@ -0,0 +1,179 @@
+<?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\ClientInterface;
+
+/**
+ * @group commands
+ * @group realm-stream
+ */
+class XCLAIM_Test extends PredisCommandTestCase
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function getExpectedCommand(): string
+    {
+        return 'Predis\Command\Redis\XCLAIM';
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function getExpectedId(): string
+    {
+        return 'XCLAIM';
+    }
+
+    /**
+     * @dataProvider argumentsProvider
+     * @group disconnected
+     */
+    public function testFilterArguments(array $actualArguments, array $expectedArguments): void
+    {
+        $command = $this->getCommand();
+        $command->setArguments($actualArguments);
+
+        $this->assertSame($expectedArguments, $command->getArguments());
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testParseResponse(): void
+    {
+        $command = $this->getCommand();
+
+        $raw = [['1-1', ['key1', 'val1']], ['2-1', ['key2', 'val2']]];
+        $expected = ['1-1' => ['key1' => 'val1'], '2-1' => ['key2' => 'val2']];
+        $this->assertSame($expected, $command->parseResponse($raw));
+        $this->assertSame($expected, $command->parseResp3Response($raw));
+
+        // JUSTID format
+        $raw = ['1-1', '2-1'];
+        $expected = ['1-1', '2-1'];
+        $this->assertSame($expected, $command->parseResponse($raw));
+        $this->assertSame($expected, $command->parseResp3Response($raw));
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testPrefixKeys(): void
+    {
+        $arguments = ['stream', 'group', 'consumer', 0, 'id1'];
+        $expected = ['prefix:stream', 'group', 'consumer', 0, 'id1'];
+
+        $command = $this->getCommandWithArgumentsArray($arguments);
+        $command->prefixKeys('prefix:');
+
+        $this->assertSame($expected, $command->getArguments());
+    }
+
+    /**
+     * @group connected
+     * @requiresRedisVersion >= 5.0.0
+     */
+    public function testClaim(): void
+    {
+        $redis = $this->getClient();
+        $this->testClaimWithClient($redis);
+    }
+
+    /**
+     * @group connected
+     * @requiresRedisVersion >= 5.0.0
+     */
+    public function testClaimResp3(): void
+    {
+        $redis = $this->getResp3Client();
+        $this->testClaimWithClient($redis);
+    }
+
+    private function testClaimWithClient(ClientInterface $redis): void
+    {
+        $redis->xadd('stream', ['key0' => 'val0'], '0-1');
+        $redis->xadd('stream', ['key1' => 'val1'], '1-1');
+        $redis->xadd('stream', ['key2' => 'val2'], '2-1');
+        $redis->xadd('stream', ['key3' => 'val3'], '3-1');
+
+        $redis->xgroup->create('stream', 'group', '0');
+
+        $redis->xreadgroup('group', 'consumer1', 4, null, false, 'stream', '>');
+
+        // Claim one
+        $claimed = $redis->xclaim('stream', 'group', 'consumer2', 0, '0-1');
+        $this->assertSame(['0-1' => ['key0' => 'val0']], $claimed);
+
+        // Claim many
+        $claimed = $redis->xclaim('stream', 'group', 'consumer2', 0, ['1-1', '2-1']);
+        $this->assertSame(['1-1' => ['key1' => 'val1'], '2-1' => ['key2' => 'val2']], $claimed);
+
+        // Claim deleted
+        $redis->xdel('stream', '1-1');
+        $claimed = $redis->xclaim('stream', 'group', 'consumer3', 0, ['0-1', '1-1', '2-1'], null, null, null, false, true);
+        $this->assertSame(['0-1', '2-1'], $claimed);
+
+        // Claim with all options
+        $redis->xdel('stream', '1-1');
+        $claimed = $redis->xclaim('stream', 'group', 'consumer3', 0, '3-1', 10, 100, 5, true, true, '3-1');
+        $this->assertSame(['3-1'], $claimed);
+
+        // Claim unknown
+        $claimed = $redis->xclaim('stream', 'group', 'consumer3', 0, ['4-1']);
+        $this->assertSame([], $claimed);
+    }
+
+    public function argumentsProvider(): array
+    {
+        return [
+            'with default arguments' => [
+                ['stream', 'group', 'consumer', 0, 'id1'],
+                ['stream', 'group', 'consumer', 0, 'id1'],
+            ],
+            'with array ids' => [
+                ['stream', 'group', 'consumer', 0, ['id1', 'id2']],
+                ['stream', 'group', 'consumer', 0, 'id1', 'id2'],
+            ],
+            'with IDLE modifier' => [
+                ['stream', 'group', 'consumer', 0, ['id1', 'id2'], 10],
+                ['stream', 'group', 'consumer', 0, 'id1', 'id2', 'IDLE', 10],
+            ],
+            'with TIME modifier' => [
+                ['stream', 'group', 'consumer', 0, ['id1', 'id2'], null, 12345],
+                ['stream', 'group', 'consumer', 0, 'id1', 'id2', 'TIME', 12345],
+            ],
+            'with RETRYCOUNT modifier' => [
+                ['stream', 'group', 'consumer', 0, ['id1', 'id2'], null, null, 5],
+                ['stream', 'group', 'consumer', 0, 'id1', 'id2', 'RETRYCOUNT', 5],
+            ],
+            'with FORCE modifier' => [
+                ['stream', 'group', 'consumer', 0, ['id1', 'id2'], null, null, null, true],
+                ['stream', 'group', 'consumer', 0, 'id1', 'id2', 'FORCE'],
+            ],
+            'with JUSTID modifier' => [
+                ['stream', 'group', 'consumer', 0, ['id1', 'id2'], null, null, null, false, true],
+                ['stream', 'group', 'consumer', 0, 'id1', 'id2', 'JUSTID'],
+            ],
+            'with LASTID modifier' => [
+                ['stream', 'group', 'consumer', 0, ['id1', 'id2'], null, null, null, false, false, '1-1'],
+                ['stream', 'group', 'consumer', 0, 'id1', 'id2', 'LASTID', '1-1'],
+            ],
+            'with all arguments' => [
+                ['stream', 'group', 'consumer', 100, ['id1', 'id2'], 10, 12345, 5, true, true, '1-1'],
+                ['stream', 'group', 'consumer', 100, 'id1', 'id2', 'IDLE', 10, 'TIME', 12345, 'RETRYCOUNT', 5, 'FORCE', 'JUSTID', 'LASTID', '1-1'],
+            ],
+        ];
+    }
+}