Added new stream commands, extended existing ones (#1580)

* Added new stream commands, extended existing ones

* updated the changelog

* fixed linter errors

* fixed regressions

* more linter fixes

* XADD fix

* lint fix

* Update CHANGELOG.md

* update interfaces, address PR comments

* fixed linter errors

* fixed xdelex return type

* updated new command signatures

---------

Co-authored-by: Till Krüss <tillkruss@users.noreply.github.com>

CHANGELOG.md

@@ -2,10 +2,11 @@
 
 ### Unreleased
 ### Added
+- Added support for `XDELEX` and `XACKDEL` (#1580)
 - Added missing VSIM argument (#1582)
 
 ### Changed
-### Fixed
+- Extended `XTRIM` and `XADD` commands with new parameters (#1580)
 
 ## v3.1.0 (2025-07-22)
 ### Added

src/ClientContextInterface.php

@@ -285,10 +285,12 @@ 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 xackdel(string $key, string $group, string $mode, array $ids)
  * @method $this xadd(string $key, array $dictionary, string $id = '*', array $options = null)
  * @method $this xautoclaim(string $key, string $group, string $consumer, int $minIdleTime, string $start, ?int $count = null, bool $justId = false)
  * @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 xdel(string $key, string ...$id)
+ * @method $this xdelex(string $key, string $mode, array $ids)
  * @method $this xlen(string $key)
  * @method $this xpending(string $key, string $group, ?int $minIdleTime = null, ?string $start = null, ?string $end = null, ?int $count = null, ?string $consumer = null)
  * @method $this xrevrange(string $key, string $end, string $start, ?int $count = null)

src/ClientInterface.php

@@ -296,10 +296,12 @@ use Predis\Response\Status;
  * @method array             tsrange(string $key, $fromTimestamp, $toTimestamp, ?RangeArguments $arguments = null)
  * @method array             tsrevrange(string $key, $fromTimestamp, $toTimestamp, ?RangeArguments $arguments = null)
  * @method int               xack(string $key, string $group, string ...$id)
+ * @method array             xackdel(string $key, string $group, string $mode, array $ids)
  * @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 array             xdelex(string $key, string $mode, array $ids)
  * @method int               xlen(string $key)
  * @method array             xpending(string $key, string $group, ?int $minIdleTime = null, ?string $start = null, ?string $end = null, ?int $count = null, ?string $consumer = null)
  * @method array             xrevrange(string $key, string $end, string $start, ?int $count = null)

src/Command/Redis/XACKDEL.php

@@ -0,0 +1,46 @@
+<?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;
+
+/**
+ * @see http://redis.io/commands/xackdel
+ */
+class XACKDEL extends RedisCommand
+{
+    /**
+     * {@inheritdoc}
+     */
+    public function getId()
+    {
+        return 'XACKDEL';
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function setArguments(array $arguments)
+    {
+        $processedArguments = [$arguments[0], $arguments[1], strtoupper($arguments[2])];
+
+        array_push($processedArguments, 'IDS', strval(count($arguments[3])), ...$arguments[3]);
+
+        parent::setArguments($processedArguments);
+    }
+
+    public function prefixKeys($prefix)
+    {
+        $this->applyPrefixForFirstArgument($prefix);
+    }
+}

src/Command/Redis/XADD.php

@@ -50,8 +50,13 @@ class XADD extends RedisCommand
             }
         }
 
+        if (isset($options['trimming'])) {
+            $args[] = strtoupper($options['trimming']);
+        }
+
         // ID, default to * to let Redis set it
         $args[] = $arguments[2] ?? '*';
+
         if (isset($arguments[1]) && is_array($arguments[1])) {
             foreach ($arguments[1] as $key => $val) {
                 $args[] = $key;

src/Command/Redis/XDELEX.php

@@ -0,0 +1,46 @@
+<?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;
+
+/**
+ * @see http://redis.io/commands/xdelex
+ */
+class XDELEX extends RedisCommand
+{
+    /**
+     * {@inheritdoc}
+     */
+    public function getId()
+    {
+        return 'XDELEX';
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function setArguments(array $arguments)
+    {
+        $processedArguments = [$arguments[0], strtoupper($arguments[1])];
+
+        array_push($processedArguments, 'IDS', strval(count($arguments[2])), ...$arguments[2]);
+
+        parent::setArguments($processedArguments);
+    }
+
+    public function prefixKeys($prefix)
+    {
+        $this->applyPrefixForFirstArgument($prefix);
+    }
+}

src/Command/Redis/XTRIM.php

@@ -49,6 +49,10 @@ class XTRIM extends RedisCommand
             $args[] = $options['limit'];
         }
 
+        if (isset($options['trimming'])) {
+            $args[] = strtoupper($options['trimming']);
+        }
+
         parent::setArguments($args);
     }
 

tests/Predis/Command/Redis/XACKDEL_Test.php

@@ -0,0 +1,134 @@
+<?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;
+
+/**
+ * @group commands
+ * @group realm-stream
+ */
+class XACKDEL_Test extends PredisCommandTestCase
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function getExpectedCommand(): string
+    {
+        return 'Predis\Command\Redis\XACKDEL';
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function getExpectedId(): string
+    {
+        return 'XACKDEL';
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testFilterArgumentsWithKEEPREF(): void
+    {
+        $arguments = ['stream', 'group1', 'KEEPREF', ['id1', 'id2']];
+        $expected = ['stream', 'group1', 'KEEPREF', 'IDS', '2', 'id1', 'id2'];
+
+        $command = $this->getCommand();
+        $command->setArguments($arguments);
+
+        $this->assertSame($expected, $command->getArguments());
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testFilterArgumentsWithDELREF(): void
+    {
+        $arguments = ['stream', 'group1', 'DELREF', ['id1', 'id2', 'id3']];
+        $expected = ['stream', 'group1', 'DELREF', 'IDS', '3', 'id1', 'id2', 'id3'];
+
+        $command = $this->getCommand();
+        $command->setArguments($arguments);
+
+        $this->assertSame($expected, $command->getArguments());
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testFilterArgumentsWithACKED(): void
+    {
+        $arguments = ['stream', 'group1', 'ACKED', ['id1']];
+        $expected = ['stream', 'group1', 'ACKED', 'IDS', '1', 'id1'];
+
+        $command = $this->getCommand();
+        $command->setArguments($arguments);
+
+        $this->assertSame($expected, $command->getArguments());
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testParseResponse(): void
+    {
+        $this->assertSame([1, 2, -1], $this->getCommand()->parseResponse([1, 2, -1]));
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testPrefixKeys(): void
+    {
+        /** @var PrefixableCommand $command */
+        $command = $this->getCommand();
+        $actualArguments = ['stream', 'group1', 'DELREF', ['id1', 'id2']];
+        $prefix = 'prefix:';
+        $expectedArguments = ['prefix:stream', 'group1', 'DELREF', 'IDS', '2', 'id1', 'id2'];
+
+        $command->setArguments($actualArguments);
+        $command->prefixKeys($prefix);
+
+        $this->assertSame($expectedArguments, $command->getArguments());
+    }
+
+    /**
+     * @group connected
+     * @requiresRedisVersion >= 8.2.0
+     */
+    public function testAcknowledgesAndDeletesSpecifiedMembers(): void
+    {
+        $redis = $this->getClient();
+
+        $redis->xadd('teststream', ['key0' => 'val0'], '0-1');
+        $redis->xadd('teststream', ['key1' => 'val1'], '1-1');
+        $redis->xadd('teststream', ['key2' => 'val2'], '2-1');
+
+        $redis->xgroup->create('teststream', 'testgroup', '0');
+
+        $redis->xreadgroup('testgroup', 'consumer1', 2, null, false, 'teststream', '>');
+
+        $result = $redis->xackdel('teststream', 'testgroup', 'KEEPREF', ['0-1', '1-1']);
+
+        $this->assertIsArray($result);
+        $this->assertCount(2, $result);
+
+        $redis->xadd('teststream', ['key3' => 'val3'], '3-1');
+        $redis->xreadgroup('testgroup', 'consumer1', 1, null, false, 'teststream', '>');
+        $result2 = $redis->xackdel('teststream', 'testgroup', 'KEEPREF', ['3-1']);
+
+        $this->assertIsArray($result2);
+        $this->assertCount(1, $result2);
+    }
+}

tests/Predis/Command/Redis/XADD_Test.php

@@ -112,6 +112,26 @@ class XADD_Test extends PredisCommandTestCase
             ['stream', '*', 'key', 'val'],
         ];
 
+        $data[] = [
+            ['stream', ['key' => 'val'], '2-3', ['trimming' => 'KEEPREF']],
+            ['stream', 'KEEPREF', '2-3', 'key', 'val'],
+        ];
+
+        $data[] = [
+            ['stream', ['key' => 'val'], '*', ['trimming' => 'KEEPREF']],
+            ['stream', 'KEEPREF', '*', 'key', 'val'],
+        ];
+
+        $data[] = [
+            [
+                'stream',
+                ['key' => 'val'],
+                '*',
+                ['trim' => ['MINID', '~', '0-1'], 'limit' => 5, 'nomkstream' => true, 'trimming' => 'KEEPREF'],
+            ],
+            ['stream', 'NOMKSTREAM', 'MINID', '~', '0-1', 'LIMIT', 5, 'KEEPREF', '*', 'key', 'val'],
+        ];
+
         return $data;
     }
 

tests/Predis/Command/Redis/XDELEX_Test.php

@@ -0,0 +1,132 @@
+<?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;
+
+/**
+ * @group commands
+ * @group realm-stream
+ */
+class XDELEX_Test extends PredisCommandTestCase
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function getExpectedCommand(): string
+    {
+        return 'Predis\Command\Redis\XDELEX';
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function getExpectedId(): string
+    {
+        return 'XDELEX';
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testFilterArgumentsWithKEEPREF(): void
+    {
+        $arguments = ['stream', 'KEEPREF', ['id1', 'id2']];
+        $expected = ['stream', 'KEEPREF', 'IDS', '2', 'id1', 'id2'];
+
+        $command = $this->getCommand();
+        $command->setArguments($arguments);
+
+        $this->assertSame($expected, $command->getArguments());
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testFilterArgumentsWithDELREF(): void
+    {
+        $arguments = ['stream', 'DELREF', ['id1', 'id2', 'id3']];
+        $expected = ['stream', 'DELREF', 'IDS', '3', 'id1', 'id2', 'id3'];
+
+        $command = $this->getCommand();
+        $command->setArguments($arguments);
+
+        $this->assertSame($expected, $command->getArguments());
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testFilterArgumentsWithACKED(): void
+    {
+        $arguments = ['stream', 'ACKED', ['id1']];
+        $expected = ['stream', 'ACKED', 'IDS', '1', 'id1'];
+
+        $command = $this->getCommand();
+        $command->setArguments($arguments);
+
+        $this->assertSame($expected, $command->getArguments());
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testParseResponse(): void
+    {
+        $this->assertSame([1, 2, -1], $this->getCommand()->parseResponse([1, 2, -1]));
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testPrefixKeys(): void
+    {
+        /** @var PrefixableCommand $command */
+        $command = $this->getCommand();
+        $actualArguments = ['stream', 'DELREF', ['id1', 'id2']];
+        $prefix = 'prefix:';
+        $expectedArguments = ['prefix:stream', 'DELREF', 'IDS', '2', 'id1', 'id2'];
+
+        $command->setArguments($actualArguments);
+        $command->prefixKeys($prefix);
+
+        $this->assertSame($expectedArguments, $command->getArguments());
+    }
+
+    /**
+     * @group connected
+     * @requiresRedisVersion >= 8.2.0
+     */
+    public function testRemovesSpecifiedMembersWithXdelex(): void
+    {
+        $redis = $this->getClient();
+
+        $redis->xadd('teststream', ['key0' => 'val0'], '0-1');
+        $redis->xadd('teststream', ['key1' => 'val1'], '1-1');
+        $redis->xadd('teststream', ['key2' => 'val2'], '2-1');
+
+        $result = $redis->xdelex('teststream', 'KEEPREF', ['0-1', '2-1']);
+
+        $this->assertIsArray($result);
+        $this->assertCount(2, $result);
+
+        $remaining = $redis->xrange('teststream', '-', '+');
+        $this->assertSame(['1-1' => ['key1' => 'val1']], $remaining);
+
+        $redis->xadd('teststream', ['key3' => 'val3'], '3-1');
+        $result2 = $redis->xdelex('teststream', 'KEEPREF', ['3-1']);
+
+        $this->assertIsArray($result2);
+        $this->assertCount(1, $result2);
+    }
+}

tests/Predis/Command/Redis/XTRIM_Test.php

@@ -84,6 +84,18 @@ class XTRIM_Test extends PredisCommandTestCase
                 ['stream', 'MINID', '0-1'],
                 ['stream', 'MINID', '0-1'],
             ],
+            [
+                ['stream', 'MINID', '0-1', ['trimming' => 'KEEPREF']],
+                ['stream', 'MINID', '0-1', 'KEEPREF'],
+            ],
+            [
+                ['stream', 'MINID', '0-1', ['limit' => 10, 'trimming' => 'KEEPREF']],
+                ['stream', 'MINID', '0-1', 'LIMIT', 10, 'KEEPREF'],
+            ],
+            [
+                ['stream', ['MINID'], '0-1', ['limit' => 10, 'trimming' => 'ACKED']],
+                ['stream', 'MINID', '0-1', 'LIMIT', 10, 'ACKED'],
+            ],
         ];
     }