geckon01/predis
1
0
Fork 0
mirror of https://github.com/predis/predis.git synced 2026-09-12 03:17:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Added support for idempotent stream API (#1632)

Browse Source 
* Added augment commands support

* Added support for Redis 8.6

* Codestyle changes and test fixes

* Fixed hybdrid tests

* Updated CHANGELOG.md and composer.json

* Fixed XINFO tests

* Added support for idempontent stream API

* Update CHANGELOG.md

* Updated expected exported files

* Removed agentic files

* Removed .augment entry
This commit is contained in:
Vladyslav Vildanov
2026-02-09 10:25:51 +02:00
committed by GitHub
parent 861bde010a
commit e851ed6ceb
7 changed files with 324 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+1
View File
@@ -10,6 +10,7 @@
### Added
- Added retry support (#1616)
- Added support for VRANGE command (#1623)
- Added support for idempotent stream API (#1632)
- Added support for HOTKEYS container command (#1630)
### Maintenance
src/ClientContextInterface.php
+1
View File
@@ -295,6 +295,7 @@ use Predis\Command\Redis\VADD;
* @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 xcfgset(string $key, ?int $duration = null, ?int $maxsize = null)
* @method $this xdel(string $key, string ...$id)
* @method $this xdelex(string $key, string $mode, array $ids)
* @method $this xlen(string $key)
src/ClientInterface.php
+1
View File
@@ -306,6 +306,7 @@ use Predis\Response\Status;
* @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 Status xcfgset(string $key, ?int $duration = null, ?int $maxsize = null)
* @method int xdel(string $key, string ...$id)
* @method array xdelex(string $key, string $mode, array $ids)
* @method int xlen(string $key)
src/Command/Redis/XADD.php
+14 -4
View File
@@ -41,6 +41,20 @@ class XADD extends RedisCommand
$args[] = 'NOMKSTREAM';
}
if (isset($options['trimming'])) {
$args[] = strtoupper($options['trimming']);
}
// IDMPAUTO or IDMP options (mutually exclusive)
if (isset($options['idmpauto'])) {
$args[] = 'IDMPAUTO';
$args[] = $options['idmpauto'];
} elseif (isset($options['idmp']) && is_array($options['idmp'])) {
$args[] = 'IDMP';
$args[] = $options['idmp'][0]; // pid
$args[] = $options['idmp'][1]; // iid
}
if (isset($options['trim']) && is_array($options['trim'])) {
array_push($args, ...$options['trim']);
@@ -50,10 +64,6 @@ class XADD extends RedisCommand
}
}
if (isset($options['trimming'])) {
$args[] = strtoupper($options['trimming']);
}
// ID, default to * to let Redis set it
$args[] = $arguments[2] ?? '*';
src/Command/Redis/XCFGSET.php
+58
View File
@@ -0,0 +1,58 @@
<?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/xcfgset
*
* XCFGSET key [IDMP-DURATION duration] [IDMP-MAXSIZE maxsize]
*
* Configures the idempotency parameters for a stream's IDMP map.
*/
class XCFGSET extends RedisCommand
{
/**
* {@inheritdoc}
*/
public function getId()
{
return 'XCFGSET';
}
/**
* {@inheritdoc}
*/
public function setArguments(array $arguments)
{
$processedArguments = [$arguments[0]];
// IDMP-DURATION option
if (isset($arguments[1]) && $arguments[1] !== null) {
array_push($processedArguments, 'IDMP-DURATION', $arguments[1]);
}
// IDMP-MAXSIZE option
if (isset($arguments[2]) && $arguments[2] !== null) {
array_push($processedArguments, 'IDMP-MAXSIZE', $arguments[2]);
}
parent::setArguments($processedArguments);
}
public function prefixKeys($prefix)
{
$this->applyPrefixForFirstArgument($prefix);
}
}
tests/Predis/Command/Redis/XADD_Test.php
+83 -1
View File
@@ -129,7 +129,36 @@ class XADD_Test extends PredisCommandTestCase
'*',
['trim' => ['MINID', '~', '0-1'], 'limit' => 5, 'nomkstream' => true, 'trimming' => 'KEEPREF'],
],
['stream', 'NOMKSTREAM', 'MINID', '~', '0-1', 'LIMIT', 5, 'KEEPREF', '*', 'key', 'val'],
['stream', 'NOMKSTREAM', 'KEEPREF', 'MINID', '~', '0-1', 'LIMIT', 5, '*', 'key', 'val'],
];
// Test IDMPAUTO option
$data[] = [
['stream', ['key' => 'val'], '*', ['idmpauto' => 'producer1']],
['stream', 'IDMPAUTO', 'producer1', '*', 'key', 'val'],
];
// Test IDMP option
$data[] = [
['stream', ['key' => 'val'], '*', ['idmp' => ['producer1', '42']]],
['stream', 'IDMP', 'producer1', '42', '*', 'key', 'val'],
];
// Test IDMPAUTO with trimming
$data[] = [
['stream', ['key' => 'val'], '*', ['idmpauto' => 'producer1', 'trimming' => 'KEEPREF']],
['stream', 'KEEPREF', 'IDMPAUTO', 'producer1', '*', 'key', 'val'],
];
// Test IDMP with trim and nomkstream
$data[] = [
[
'stream',
['key' => 'val'],
'*',
['idmp' => ['producer1', '42'], 'trim' => ['MAXLEN', '~', '100'], 'nomkstream' => true],
],
['stream', 'NOMKSTREAM', 'IDMP', 'producer1', '42', 'MAXLEN', '~', '100', '*', 'key', 'val'],
];
return $data;
@@ -340,4 +369,57 @@ class XADD_Test extends PredisCommandTestCase
$redis->set('foo', 'bar');
$redis->xadd('foo', ['key' => 'val']);
}
/**
* @group connected
* @requiresRedisVersion >= 8.5.0
*/
public function testIdmpautoGeneratesSameIdForSameContent(): void
{
$redis = $this->getClient();
// Add the same message twice with IDMPAUTO
$id1 = $redis->xadd('stream', ['field1' => 'value1', 'field2' => 'value2'], '*', ['idmpauto' => 'producer1']);
$id2 = $redis->xadd('stream', ['field1' => 'value1', 'field2' => 'value2'], '*', ['idmpauto' => 'producer1']);
// Both should return the same ID (idempotent)
$this->assertSame($id1, $id2);
// Only one entry should exist in the stream
$this->assertSame(1, $redis->xlen('stream'));
}
/**
* @group connected
* @requiresRedisVersion >= 8.5.0
*/
public function testIdmpUsesSameIdForSameIid(): void
{
$redis = $this->getClient();
// Add the same message twice with IDMP using the same iid
$id1 = $redis->xadd('stream', ['field1' => 'value1', 'field2' => 'value2'], '*', ['idmp' => ['producer1', '42']]);
$id2 = $redis->xadd('stream', ['field1' => 'value1', 'field2' => 'value2'], '*', ['idmp' => ['producer1', '42']]);
// Both should return the same ID (idempotent)
$this->assertSame($id1, $id2);
// Only one entry should exist in the stream
$this->assertSame(1, $redis->xlen('stream'));
}
/**
* @group connected
* @requiresRedisVersion >= 8.5.0
*/
public function testIdmpWithDifferentIidCreatesNewEntry(): void
{
$redis = $this->getClient();
// Add messages with different iids
$id1 = $redis->xadd('stream', ['field1' => 'value1'], '*', ['idmp' => ['producer1', '42']]);
$id2 = $redis->xadd('stream', ['field1' => 'value1'], '*', ['idmp' => ['producer1', '43']]);
// Should create two different entries
$this->assertNotSame($id1, $id2);
$this->assertSame(2, $redis->xlen('stream'));
}
}
tests/Predis/Command/Redis/XCFGSET_Test.php
+166
View File
@@ -0,0 +1,166 @@
<?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;
/**
* @group commands
* @group realm-stream
*/
class XCFGSET_Test extends PredisCommandTestCase
{
/**
* {@inheritdoc}
*/
protected function getExpectedCommand(): string
{
return 'Predis\Command\Redis\XCFGSET';
}
/**
* {@inheritdoc}
*/
protected function getExpectedId(): string
{
return 'XCFGSET';
}
/**
* @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
{
$this->assertSame('OK', $this->getCommand()->parseResponse('OK'));
}
/**
* @group disconnected
*/
public function testPrefixKeys(): void
{
$arguments = ['stream', 100, 1000];
$expected = ['prefix:stream', 'IDMP-DURATION', 100, 'IDMP-MAXSIZE', 1000];
$command = $this->getCommandWithArgumentsArray($arguments);
$command->prefixKeys('prefix:');
$this->assertSame($expected, $command->getArguments());
}
/**
* @group connected
* @requiresRedisVersion >= 8.5.0
*/
public function testConfigureStreamIdempotencyParameters(): void
{
$redis = $this->getClient();
// Create a stream first
$redis->xadd('stream', ['field' => 'value']);
// Configure IDMP parameters
$response = $redis->xcfgset('stream', 100, 1000);
$this->assertEquals('OK', $response);
}
/**
* @group connected
* @requiresRedisVersion >= 8.5.0
*/
public function testIIDsAreFlushedAfterDuration(): void
{
$redis = $this->getClient();
// Configure stream with very short IDMP-DURATION (2 seconds)
$redis->xadd('stream', ['field' => 'value']);
$redis->xcfgset('stream', 1, 1000);
// Add message with IID using IDMP
$id1 = $redis->xadd('stream', ['field' => 'value1'], '*', ['idmp' => ['producer1', 'iid-1']]);
// Try to add duplicate immediately - should return same ID (idempotent)
$id2 = $redis->xadd('stream', ['field' => 'value2'], '*', ['idmp' => ['producer1', 'iid-1']]);
$this->assertSame($id1, $id2);
// Wait for duration to expire (2 seconds + buffer)
sleep(2);
// Now the same IID should create a new entry (IID was flushed)
$id3 = $redis->xadd('stream', ['field' => 'value3'], '*', ['idmp' => ['producer1', 'iid-1']]);
$this->assertNotSame($id1, $id3);
}
/**
* @group connected
* @requiresRedisVersion >= 8.5.0
*/
public function testIIDsAreFlushedWhenExceedingMaxsize(): void
{
$redis = $this->getClient();
// Configure stream with small IDMP-MAXSIZE (3 IIDs)
$redis->xadd('stream', ['field' => 'value']);
$redis->xcfgset('stream', 100, 3);
// Add 3 messages with different IIDs (fills the IDMP map)
$id1 = $redis->xadd('stream', ['field' => 'value1'], '*', ['idmp' => ['producer1', 'iid-1']]);
$id2 = $redis->xadd('stream', ['field' => 'value2'], '*', ['idmp' => ['producer1', 'iid-2']]);
$id3 = $redis->xadd('stream', ['field' => 'value3'], '*', ['idmp' => ['producer1', 'iid-3']]);
// Try to add duplicate of iid-1 - should return same ID (idempotent)
$id1Dup = $redis->xadd('stream', ['field' => 'duplicate'], '*', ['idmp' => ['producer1', 'iid-1']]);
$this->assertSame($id1, $id1Dup);
// Add a 4th unique IID - this should cause oldest IID (iid-1) to be flushed
$id4 = $redis->xadd('stream', ['field' => 'value4'], '*', ['idmp' => ['producer1', 'iid-4']]);
$this->assertNotNull($id4);
// Now iid-1 should create a new entry (it was flushed due to maxsize)
$id1New = $redis->xadd('stream', ['field' => 'value5'], '*', ['idmp' => ['producer1', 'iid-1']]);
$this->assertNotSame($id1, $id1New);
}
public function argumentsProvider(): array
{
return [
'with key only' => [
['stream'],
['stream'],
],
'with IDMP-DURATION only' => [
['stream', 100],
['stream', 'IDMP-DURATION', 100],
],
'with IDMP-MAXSIZE only' => [
['stream', null, 1000],
['stream', 'IDMP-MAXSIZE', 1000],
],
'with both IDMP-DURATION and IDMP-MAXSIZE' => [
['stream', 100, 1000],
['stream', 'IDMP-DURATION', 100, 'IDMP-MAXSIZE', 1000],
],
];
}
}