diff --git a/src/ClientInterface.php b/src/ClientInterface.php index 49f8c3c6..c1362d0c 100644 --- a/src/ClientInterface.php +++ b/src/ClientInterface.php @@ -45,6 +45,7 @@ use Predis\Command\Container\Json\JSONDEBUG; use Predis\Command\Container\Search\FTCONFIG; use Predis\Command\Container\Search\FTCURSOR; use Predis\Command\Container\XGROUP; +use Predis\Command\Container\XINFO; use Predis\Command\FactoryInterface; use Predis\Configuration\OptionsInterface; use Predis\Connection\ConnectionInterface; @@ -371,6 +372,7 @@ use Predis\Response\Status; * @property JSONDEBUG $jsondebug * @property ACL $acl * @property XGROUP $xgroup + * @property XINFO $xinfo */ interface ClientInterface { diff --git a/src/Command/Argument/Stream/XInfoStreamOptions.php b/src/Command/Argument/Stream/XInfoStreamOptions.php new file mode 100644 index 00000000..71967901 --- /dev/null +++ b/src/Command/Argument/Stream/XInfoStreamOptions.php @@ -0,0 +1,49 @@ +options[] = 'FULL'; + + if (null !== $count) { + array_push($this->options, 'COUNT', $count); + } + + return $this; + } + + /** + * {@inheritDoc} + */ + public function toArray(): array + { + return $this->options; + } +} diff --git a/src/Command/Container/XINFO.php b/src/Command/Container/XINFO.php new file mode 100644 index 00000000..6f145c60 --- /dev/null +++ b/src/Command/Container/XINFO.php @@ -0,0 +1,28 @@ +setStreamArguments($arguments); + } else { + parent::setArguments($arguments); + } + } + + /** + * @param array $arguments + * @return void + */ + private function setStreamArguments(array $arguments): void + { + $processedArguments = [$arguments[0], $arguments[1]]; + + if (array_key_exists(2, $arguments) && $arguments[2] instanceof ArrayableArgument) { + $processedArguments = array_merge($processedArguments, $arguments[2]->toArray()); + } + + parent::setArguments($processedArguments); + } + + public function parseResponse($data) + { + $result = []; + + for ($i = 0, $iMax = count($data); $i < $iMax; $i++) { + if (is_array($data[$i])) { + $result[$i] = $this->parseResponse($data[$i]); + } + + if (array_key_exists($i + 1, $data)) { + if (is_array($data[$i + 1])) { + $result[$data[$i]] = $this->parseResponse($data[++$i]); + } else { + $result[$data[$i]] = $data[++$i]; + } + } + } + + return $result; + } +} diff --git a/tests/Predis/Command/Redis/XINFO_Test.php b/tests/Predis/Command/Redis/XINFO_Test.php new file mode 100644 index 00000000..bb19ff3a --- /dev/null +++ b/tests/Predis/Command/Redis/XINFO_Test.php @@ -0,0 +1,210 @@ +getCommand(); + $command->setArguments($arguments); + + $this->assertSameValues($expected, $command->getArguments()); + } + + /** + * @group disconnected + */ + public function testGroupsFilterArguments(): void + { + $arguments = ['GROUPS', 'key']; + $expected = ['GROUPS', 'key']; + + $command = $this->getCommand(); + $command->setArguments($arguments); + + $this->assertSameValues($expected, $command->getArguments()); + } + + /** + * @dataProvider streamArgumentsProvider + * @group disconnected + */ + public function testStreamFilterArguments(array $actualArguments, array $expectedResponse): void + { + $command = $this->getCommand(); + $command->setArguments($actualArguments); + + $this->assertSameValues($expectedResponse, $command->getArguments()); + } + + /** + * @dataProvider responseProvider + * @group disconnected + */ + public function testParseResponse(array $arguments, array $actualResponse, array $expectedResponse): void + { + $command = $this->getCommand(); + $command->setArguments($arguments); + + $this->assertSame($expectedResponse, $command->parseResponse($actualResponse)); + } + + /** + * @group connected + * @group relay-incompatible + * @return void + * @requiresRedisVersion >= 6.2.0 + */ + public function testReturnsConsumersOfGivenGroup(): void + { + $redis = $this->getClient(); + + $entityId = $redis->xadd('stream', ['field' => 'value']); + + $this->assertEquals('OK', $redis->xgroup->create('stream', 'group', $entityId)); + $this->assertSame(1, $redis->xgroup->createConsumer('stream', 'group', 'consumer')); + + $response = $redis->xinfo->consumers('stream', 'group'); + + foreach ($response as $consumer) { + foreach (['name', 'pending', 'idle'] as $key) { + $this->assertArrayHasKey($key, $consumer); + } + } + } + + /** + * @group connected + * @group relay-incompatible + * @return void + * @requiresRedisVersion >= 7.0.0 + */ + public function testReturnsConsumerGroupsOfGivenStream(): void + { + $redis = $this->getClient(); + + $entityId = $redis->xadd('stream', ['field' => 'value']); + + $this->assertEquals('OK', $redis->xgroup->create('stream', 'group', $entityId)); + + $expectedResponse = [ + [ + 'name' => 'group', + 'consumers' => 0, + 'pending' => 0, + 'last-delivered-id' => $entityId, + 'entries-read' => null, + 'lag' => 0, + ], + ]; + + $this->assertSame($expectedResponse, $redis->xinfo->groups('stream')); + } + + /** + * @group connected + * @group relay-incompatible + * @return void + * @requiresRedisVersion >= 7.0.0 + */ + public function testReturnsInformationAboutGivenStream(): void + { + $redis = $this->getClient(); + + $entityId = $redis->xadd('stream', ['field' => 'value']); + $expectedResponse = [ + 'length' => 1, + 'radix-tree-keys' => 1, + 'radix-tree-nodes' => 2, + 'last-generated-id' => $entityId, + 'max-deleted-entry-id' => '0-0', + 'entries-added' => 1, + 'recorded-first-entry-id' => $entityId, + 'entries' => [ + [ + $entityId => ['field' => 'value'], + ], + ], + 'groups' => [], + ]; + + $options = new XInfoStreamOptions(); + $options->full(5); + + $this->assertSame($expectedResponse, $redis->xinfo->stream('stream', $options)); + } + + public function streamArgumentsProvider(): array + { + return [ + 'with default arguments' => [ + ['STREAM', 'key'], + ['STREAM', 'key'], + ], + 'with FULL modifier - no COUNT' => [ + ['STREAM', 'key', (new XInfoStreamOptions())->full()], + ['STREAM', 'key', 'FULL'], + ], + 'with FULL modifier - with COUNT' => [ + ['STREAM', 'key', (new XInfoStreamOptions())->full(15)], + ['STREAM', 'key', 'FULL', 'COUNT', 15], + ], + ]; + } + + public function responseProvider(): array + { + return [ + 'CONSUMERS response' => [ + ['CONSUMERS'], + [['name', 'consumer', 'pending', 0, 'idle', 3, 'inactive', -1]], + [['name' => 'consumer', 'pending' => 0, 'idle' => 3, 'inactive' => -1]], + ], + 'GROUPS response' => [ + ['GROUPS'], + [['name', 'group', 'consumers', 0, 'pending', 0, 'last-delivered-id', 3]], + [['name' => 'group', 'consumers' => 0, 'pending' => 0, 'last-delivered-id' => 3]], + ], + 'STREAM response' => [ + ['STREAM', 'key'], + [['length', 1, 'entries-added', 1, 'entries', [['id', ['field', 'value']]]]], + [['length' => 1, 'entries-added' => 1, 'entries' => [['id' => ['field' => 'value']]]]], + ], + ]; + } +}