diff --git a/CHANGELOG.md b/CHANGELOG.md index 405b5d91..7e40f805 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ - Added support for new TS commands + Indonesian language support integration test (#1695) - Added support for new COLLECT reducer for FT.AGGREGATE (#1699) - Added support for `SDIFFCARD` and `SUNIONCARD` command +- Added support for `TS.READ` command ### Fixed - Fixed Sentinel does not wipe servers on exception caused (#1694) diff --git a/src/ClientContextInterface.php b/src/ClientContextInterface.php index 7cfbfb4f..503d18ea 100644 --- a/src/ClientContextInterface.php +++ b/src/ClientContextInterface.php @@ -39,6 +39,7 @@ use Predis\Command\Argument\TimeSeries\MGetArguments; use Predis\Command\Argument\TimeSeries\MRangeArguments; use Predis\Command\Argument\TimeSeries\NRangeArguments; use Predis\Command\Argument\TimeSeries\RangeArguments; +use Predis\Command\Argument\TimeSeries\ReadArguments; use Predis\Command\CommandInterface; use Predis\Command\Container\ACL; use Predis\Command\Container\CLIENT; @@ -315,6 +316,7 @@ use Predis\Command\Redis\VADD; * @method $this tsnrevrange(array $keys, $fromTimestamp, $toTimestamp, ?NRangeArguments $arguments = null) * @method $this tsqueryindex(string ...$filterExpression) * @method $this tsrange(string $key, $fromTimestamp, $toTimestamp, ?RangeArguments $arguments = null) + * @method $this tsread(string $key, $timestamp, ?ReadArguments $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) diff --git a/src/ClientInterface.php b/src/ClientInterface.php index d2042726..393f7eb6 100644 --- a/src/ClientInterface.php +++ b/src/ClientInterface.php @@ -39,6 +39,7 @@ use Predis\Command\Argument\TimeSeries\MGetArguments; use Predis\Command\Argument\TimeSeries\MRangeArguments; use Predis\Command\Argument\TimeSeries\NRangeArguments; use Predis\Command\Argument\TimeSeries\RangeArguments; +use Predis\Command\Argument\TimeSeries\ReadArguments; use Predis\Command\CommandInterface; use Predis\Command\Container\ACL; use Predis\Command\Container\CLIENT; @@ -327,6 +328,7 @@ use Predis\Response\Status; * @method array tsnrevrange(array $keys, $fromTimestamp, $toTimestamp, ?NRangeArguments $arguments = null) * @method array tsqueryindex(string ...$filterExpression) * @method array tsrange(string $key, $fromTimestamp, $toTimestamp, ?RangeArguments $arguments = null) + * @method array tsread(string $key, $timestamp, ?ReadArguments $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) diff --git a/src/Command/Argument/TimeSeries/ReadArguments.php b/src/Command/Argument/TimeSeries/ReadArguments.php new file mode 100644 index 00000000..d3843230 --- /dev/null +++ b/src/Command/Argument/TimeSeries/ReadArguments.php @@ -0,0 +1,60 @@ +arguments, 'BLOCK', $milliseconds, $minCount); + + return $this; + } + + /** + * Caps the reply at the given number of samples. When more samples qualify, + * the oldest ones are returned so callers can page forward. + * + * @param int $maxCount Reply cap, positive. + * @return $this + */ + public function maxCount(int $maxCount): self + { + array_push($this->arguments, 'MAX_COUNT', $maxCount); + + return $this; + } + + /** + * {@inheritDoc} + */ + public function toArray(): array + { + return $this->arguments; + } +} diff --git a/src/Command/Redis/TimeSeries/TSREAD.php b/src/Command/Redis/TimeSeries/TSREAD.php new file mode 100644 index 00000000..1cd9e441 --- /dev/null +++ b/src/Command/Redis/TimeSeries/TSREAD.php @@ -0,0 +1,46 @@ +toArray() : []; + + parent::setArguments(array_merge( + [$key, $timestamp], + $commandArguments + )); + } + + public function prefixKeys($prefix) + { + $this->applyPrefixForFirstArgument($prefix); + } +} diff --git a/tests/Predis/Command/Redis/TimeSeries/TSREAD_Test.php b/tests/Predis/Command/Redis/TimeSeries/TSREAD_Test.php new file mode 100644 index 00000000..024d267e --- /dev/null +++ b/tests/Predis/Command/Redis/TimeSeries/TSREAD_Test.php @@ -0,0 +1,273 @@ +getCommand(); + $command->setArguments($actualArguments); + + $this->assertSameValues($expectedArguments, $command->getArguments()); + } + + /** + * @group disconnected + */ + public function testParseResponse(): void + { + $response = [[100, '1'], [200, '2'], [300, '3']]; + + $this->assertSame($response, $this->getCommand()->parseResponse($response)); + } + + /** + * @group connected + * @return void + * @requiresRedisVersion >= 8.10.0 + */ + public function testReadsAllSamplesAtOnce(): void + { + $redis = $this->getClient(); + + $this->assertEquals('OK', $redis->tscreate('sensor:1')); + $this->assertSame(100, $redis->tsadd('sensor:1', 100, 1.0)); + $this->assertSame(200, $redis->tsadd('sensor:1', 200, 2.0)); + $this->assertSame(300, $redis->tsadd('sensor:1', 300, 3.0)); + + $this->assertEquals( + [[100, '1'], [200, '2'], [300, '3']], + $redis->tsread('sensor:1', 0) + ); + } + + /** + * @group connected + * @return void + * @requiresRedisVersion >= 8.10.0 + */ + public function testReadsSamplesUsingCursorPagination(): void + { + $redis = $this->getClient(); + + $this->assertEquals('OK', $redis->tscreate('sensor:1')); + $this->assertSame(100, $redis->tsadd('sensor:1', 100, 1.0)); + $this->assertSame(200, $redis->tsadd('sensor:1', 200, 2.0)); + $this->assertSame(300, $redis->tsadd('sensor:1', 300, 3.0)); + + $arguments = (new ReadArguments())->maxCount(2); + + // First page returns the oldest max_count samples. + $firstPage = $redis->tsread('sensor:1', 0, $arguments); + $this->assertEquals([[100, '1'], [200, '2']], $firstPage); + + // Cursor advances past the last returned timestamp. + $lastTimestamp = $firstPage[count($firstPage) - 1][0]; + $secondPage = $redis->tsread('sensor:1', $lastTimestamp + 1, $arguments); + $this->assertEquals([[300, '3']], $secondPage); + + // Cursor past the newest sample returns an empty list. + $lastTimestamp = $secondPage[count($secondPage) - 1][0]; + $this->assertEquals([], $redis->tsread('sensor:1', $lastTimestamp + 1, $arguments)); + } + + /** + * @group connected + * @return void + * @requiresRedisVersion >= 8.10.0 + */ + public function testReadsSamplesUsingSpecialTimestampCursors(): void + { + $redis = $this->getClient(); + + $this->assertEquals('OK', $redis->tscreate('sensor:1')); + $this->assertSame(100, $redis->tsadd('sensor:1', 100, 1.0)); + $this->assertSame(200, $redis->tsadd('sensor:1', 200, 2.0)); + + // "-" reads from the earliest sample. + $this->assertEquals([[100, '1'], [200, '2']], $redis->tsread('sensor:1', '-')); + + // "+" is the latest sample's timestamp, inclusive. + $this->assertEquals([[200, '2']], $redis->tsread('sensor:1', '+')); + + // "$" qualifies only samples added after the command, so without + // blocking it returns an empty list immediately. + $this->assertEquals([], $redis->tsread('sensor:1', '$')); + } + + /** + * @group connected + * @return void + * @requiresRedisVersion >= 8.10.0 + */ + public function testReadsMissingKeyReturnsEmptyListWithoutError(): void + { + $redis = $this->getClient(); + + $this->assertEquals([], $redis->tsread('missing_key', 0)); + } + + /** + * @group connected + * @return void + * @requiresRedisVersion >= 8.10.0 + */ + public function testBlocksForGivenTimeoutWhenNotEnoughSamplesAvailable(): void + { + $redis = $this->getClient(); + + $this->assertEquals('OK', $redis->tscreate('sensor:1')); + $this->assertSame(100, $redis->tsadd('sensor:1', 100, 1.0)); + $this->assertSame(200, $redis->tsadd('sensor:1', 200, 2.0)); + + // Fewer qualifying samples than min_count: the command blocks until + // the timeout elapses, then returns whatever qualifies. + $start = microtime(true); + $response = $redis->tsread('sensor:1', 0, (new ReadArguments())->block(100, 5)); + $elapsedMs = (microtime(true) - $start) * 1000; + + $this->assertGreaterThanOrEqual(100, $elapsedMs); + $this->assertEquals([[100, '1'], [200, '2']], $response); + } + + /** + * @group connected + * @return void + * @requiresRedisVersion >= 8.10.0 + */ + public function testReturnsImmediatelyWhenMinCountSamplesAlreadyAvailable(): void + { + $redis = $this->getClient(); + + $this->assertEquals('OK', $redis->tscreate('sensor:1')); + $this->assertSame(100, $redis->tsadd('sensor:1', 100, 1.0)); + $this->assertSame(200, $redis->tsadd('sensor:1', 200, 2.0)); + + // min_count is already satisfied, so BLOCK 0 (wait indefinitely) + // returns immediately, capped by max_count. + $response = $redis->tsread('sensor:1', 0, (new ReadArguments())->block(0, 2)->maxCount(2)); + + $this->assertEquals([[100, '1'], [200, '2']], $response); + } + + /** + * @group connected + * @return void + * @requiresRedisVersion >= 8.10.0 + */ + public function testReadsAllSamplesAtOnceResp3(): void + { + $redis = $this->getResp3Client(); + + $this->assertEquals('OK', $redis->tscreate('sensor:1')); + $this->assertSame(100, $redis->tsadd('sensor:1', 100, 1.0)); + $this->assertSame(200, $redis->tsadd('sensor:1', 200, 2.0)); + + $this->assertEquals( + [[100, '1'], [200, '2']], + $redis->tsread('sensor:1', 0) + ); + } + + /** + * @group connected + * @dataProvider unexpectedValuesProvider + * @param ReadArguments $arguments + * @param string $expectedExceptionMessage + * @return void + * @requiresRedisVersion >= 8.10.0 + */ + public function testThrowsExceptionOnCountConstraintViolations( + ReadArguments $arguments, + string $expectedExceptionMessage + ): void { + $redis = $this->getClient(); + + $this->assertEquals('OK', $redis->tscreate('sensor:1')); + $this->assertSame(100, $redis->tsadd('sensor:1', 100, 1.0)); + + $this->expectException(ServerException::class); + $this->expectExceptionMessage($expectedExceptionMessage); + + $redis->tsread('sensor:1', 0, $arguments); + } + + public function argumentsProvider(): array + { + return [ + 'with default arguments' => [ + ['key', 0], + ['key', 0], + ], + 'with special timestamp cursor' => [ + ['key', '$'], + ['key', '$'], + ], + 'with BLOCK modifier' => [ + ['key', 0, (new ReadArguments())->block(1000, 5)], + ['key', 0, 'BLOCK', 1000, 5], + ], + 'with MAX_COUNT modifier' => [ + ['key', 0, (new ReadArguments())->maxCount(10)], + ['key', 0, 'MAX_COUNT', 10], + ], + 'with all arguments' => [ + ['key', 0, (new ReadArguments())->block(1000, 5)->maxCount(10)], + ['key', 0, 'BLOCK', 1000, 5, 'MAX_COUNT', 10], + ], + ]; + } + + public function unexpectedValuesProvider(): array + { + return [ + 'with min_count greater than MAX_COUNT' => [ + (new ReadArguments())->block(10, 5)->maxCount(2), + 'BLOCK min_count must be <= MAX_COUNT', + ], + 'with non-positive MAX_COUNT' => [ + (new ReadArguments())->maxCount(0), + 'MAX_COUNT must be a positive integer', + ], + ]; + } +}