Merge pull request #8 from asiries335/mvp-6

Mvp 6

Author: asiries335

README.md

@@ -135,6 +135,8 @@ $collection = $client->stream('test')->get();
 
 _Listen to a stream_
 
+functional works on a package basis https://github.com/reactphp/event-loop
+
 ```php
 $client->stream('test')->listen(
     function (\Asiries335\redisSteamPhp\Data\Message $message) {
@@ -143,4 +145,34 @@ $client->stream('test')->listen(
 );
 ```
 
-functional works on a package basis https://github.com/reactphp/event-loop
+
+_Create a new consumer group_
+
+```php
+$streamName = 'test';
+$groupName  = 'demo-group-1';
+$isShowFullHistoryStream = false;
+
+// return bool or ErrorException.
+$client->streamGroupConsumer($streamName)->create($groupName, $isShowFullHistoryStream);
+```
+
+_Destroy a consumer group_
+
+```php
+$streamName = 'test';
+$groupName  = 'demo-group-1';
+
+// return bool or ErrorException.
+$client->streamGroupConsumer($streamName)->destroy($groupName);
+```
+
+_Delete a consumer from group_
+```php
+$streamName = 'test';
+$groupName  = 'demo-group-1';
+$consumerName = 'consumer-name';
+
+// return bool or ErrorException.
+$client->streamGroupConsumer($streamName)->deleteConsumer($groupName, $consumerName);
+```

src/Client.php

@@ -33,12 +33,24 @@ public function __construct(ClientRedisStreamPhpInterface $redisClient)
     /**
      * Work with stream
      *
-     * @param string $nameStream Name stream
+     * @param string $streamName Name stream
      *
-     * @return mixed
+     * @return Stream
      */
-    public function stream(string $nameStream)
+    public function stream(string $streamName) : Stream
     {
-        return new Stream($this->_client, $nameStream);
+        return new Stream($this->_client, $streamName);
+    }
+
+    /**
+     * Work with stream group
+     *
+     * @param string $streamName Name stream
+     *
+     * @return StreamGroupConsumer
+     */
+    public function streamGroupConsumer(string $streamName) : StreamGroupConsumer
+    {
+        return new StreamGroupConsumer($this->_client, $streamName);
     }
 }

src/Data/Constants.php

@@ -20,13 +20,41 @@ final class Constants
      */
     public const COMMAND_XREAD = 'xread';
 
+    /**
+     * COMMAND XGROUP
+     *
+     * @var string
+     */
+    public const COMMAND_XGROUP = 'xgroup';
+
     /**
      * COMMAND XRANGE
      *
      * @var string
      */
     public const COMMAND_XRANGE = 'xrange';
 
+    /**
+     * COMMAND OPTION CREATE
+     *
+     * @var string
+     */
+    public const COMMAND_OPTION_XGROUP_CREATE = 'CREATE';
+
+    /**
+     * COMMAND OPTION DESTROY
+     *
+     * @var string
+     */
+    public const COMMAND_OPTION_XGROUP_DESTROY = 'DESTROY';
+
+    /**
+     * COMMAND OPTION DELCONSUMER
+     *
+     * @var string
+     */
+    public const COMMAND_OPTION_XGROUP_DELCONSUMER = 'DELCONSUMER';
+
     /**
      * TIME TICK INTERVAL
      *

src/RedisStream.php

@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Asiries335\redisSteamPhp;
+
+abstract class RedisStream
+{
+    /**
+     * Client
+     *
+     * @var ClientRedisStreamPhpInterface
+     */
+    protected $_client;
+
+    /**
+     * Name stream
+     *
+     * @var string
+     */
+    protected $_streamName;
+
+    /**
+     * Stream constructor.
+     *
+     * @param ClientRedisStreamPhpInterface $client     ClientRedisInterface
+     * @param string                        $nameStream Name stream
+     */
+    public function __construct(ClientRedisStreamPhpInterface $client, string $nameStream)
+    {
+        $this->_client     = $client;
+        $this->_streamName = $nameStream;
+    }
+}

src/Stream.php

@@ -18,34 +18,8 @@
 use Asiries335\redisSteamPhp\Hydrator\MessageHydrator;
 
 
-final class Stream
+final class Stream extends RedisStream
 {
-    /**
-     * Client
-     *
-     * @var ClientRedisStreamPhpInterface
-     */
-    private $_client;
-
-    /**
-     * Name stream
-     *
-     * @var string
-     */
-    private $_streamName;
-
-    /**
-     * Stream constructor.
-     *
-     * @param ClientRedisStreamPhpInterface $client     ClientRedisInterface
-     * @param string                        $nameStream Name stream
-     */
-    public function __construct(ClientRedisStreamPhpInterface $client, string $nameStream)
-    {
-        $this->_client     = $client;
-        $this->_streamName = $nameStream;
-    }
-
     /**
      * Appends the specified stream entry to the stream at the specified key
      *

src/StreamGroupConsumer.php

@@ -0,0 +1,102 @@
+<?php
+declare(strict_types=1);
+
+namespace Asiries335\redisSteamPhp;
+
+
+use Asiries335\redisSteamPhp\Data\Constants;
+use Asiries335\redisSteamPhp\Dto\StreamCommandCallTransporter;
+
+final class StreamGroupConsumer extends RedisStream
+{
+    /**
+     * Create group Consumer
+     *
+     * @param string $groupName
+     * @param bool $isShowFullHistoryStream
+     *
+     * @return bool
+     *
+     * @throws \Exception
+     */
+    public function create(string $groupName, bool $isShowFullHistoryStream = true) : bool
+    {
+        $transporter = new StreamCommandCallTransporter(
+            [
+                'command' => Constants::COMMAND_XGROUP,
+                'args'    => [
+                    Constants::COMMAND_OPTION_XGROUP_CREATE,
+                    $this->_streamName,
+                    $groupName,
+                    (int) $isShowFullHistoryStream
+                ]
+            ]
+        );
+
+        try {
+            return (bool) $this->_client->call($transporter);
+        } catch (\Exception $exception) {
+            throw $exception;
+        }
+    }
+
+    /**
+     * Delete group Consumer
+     *
+     * @param string $groupName
+     *
+     * @return bool
+     *
+     * @throws \Exception
+     */
+    public function destroy(string $groupName) : bool
+    {
+        $transporter = new StreamCommandCallTransporter(
+            [
+                'command' => Constants::COMMAND_XGROUP,
+                'args'    => [
+                    Constants::COMMAND_OPTION_XGROUP_DESTROY,
+                    $this->_streamName,
+                    $groupName,
+                ]
+            ]
+        );
+
+        try {
+            return (bool) $this->_client->call($transporter);
+        } catch (\Exception $exception) {
+            throw $exception;
+        }
+    }
+
+    /**
+     * Delete consumer from group
+     *
+     * @param string $groupName
+     * @param string $consumerName
+     *
+     * @return bool
+     *
+     * @throws \Exception
+     */
+    public function deleteConsumer(string $groupName, string $consumerName) : bool
+    {
+        $transporter = new StreamCommandCallTransporter(
+            [
+                'command' => Constants::COMMAND_XGROUP,
+                'args'    => [
+                    Constants::COMMAND_OPTION_XGROUP_DELCONSUMER,
+                    $this->_streamName,
+                    $groupName,
+                    $consumerName
+                ]
+            ]
+        );
+
+        try {
+            return (bool) $this->_client->call($transporter);
+        } catch (\Exception $exception) {
+            throw $exception;
+        }
+    }
+}