renaming core functions

.gitattributes

@@ -5,6 +5,7 @@
 /*.neon                 export-ignore
 /*.xml                  export-ignore
 /*.xml.dist             export-ignore
+/*.lock                 export-ignore
 docker-compose.yml      export-ignore
 Makefile                export-ignore
 rector.php              export-ignore

src/Consumer.php

@@ -14,7 +14,7 @@
 use kuaukutsu\queue\core\ConsumerInterface;
 use kuaukutsu\queue\core\SchemaInterface;
 use kuaukutsu\poc\queue\stream\event\EventDispatcher;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup;
+use kuaukutsu\poc\queue\stream\internal\stream\RedisConsume;
 use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
 use kuaukutsu\poc\queue\stream\internal\workflow\TaskHandler;
 use kuaukutsu\poc\queue\stream\internal\workflow\WorkflowCatch;
@@ -50,13 +50,14 @@ public function __construct(
     #[Override]
     public function consume(SchemaInterface $schema): void
     {
-        $stream = new RedisStreamGroup($this->redis, $this->options, $schema);
+        $string = new RedisString($this->redis, $schema);
+        $stream = new RedisConsume($this->redis, $this->options, $schema);
         $stream->create();
 
         $this->ctx = new Context(
             $schema,
             $stream,
-            new RedisString($this->redis),
+            $string,
             $this->eventDispatcher,
         );
 

src/Publisher.php

@@ -13,7 +13,7 @@
 use kuaukutsu\queue\core\QueueMessage;
 use kuaukutsu\queue\core\QueueTask;
 use kuaukutsu\queue\core\SchemaInterface;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStream;
+use kuaukutsu\poc\queue\stream\internal\stream\RedisPublish;
 use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
 use kuaukutsu\poc\queue\stream\internal\Payload;
 
@@ -25,14 +25,8 @@
  */
 final readonly class Publisher implements PublisherInterface
 {
-    private RedisStream $stream;
-
-    private RedisString $string;
-
-    public function __construct(RedisClient $redis)
+    public function __construct(private RedisClient $redis)
     {
-        $this->stream = new RedisStream($redis);
-        $this->string = new RedisString($redis);
     }
 
     /**
@@ -42,15 +36,18 @@ public function __construct(RedisClient $redis)
     #[Override]
     public function push(SchemaInterface $schema, QueueTask $task, ?QueueContext $context = null): string
     {
-        $this->string->set(
+        $string = new RedisString($this->redis, $schema);
+        $stream = new RedisPublish($this->redis, $schema);
+
+        $string->set(
             $task->getUuid(),
             QueueMessage::makeMessage($task, $context ?? QueueContext::make($schema)),
         );
 
         try {
-            $this->stream->add($schema, Payload::fromTask($task)->toArray());
+            $stream->add(Payload::fromTask($task)->toArray());
         } catch (Throwable $exception) {
-            $this->string->del($task->getUuid());
+            $string->del($task->getUuid());
             throw new QueuePublishException($schema, $exception);
         }
 

src/internal/Context.php

@@ -12,7 +12,7 @@
 use kuaukutsu\poc\queue\stream\event\EventDispatcher;
 use kuaukutsu\poc\queue\stream\event\MessageAckEvent;
 use kuaukutsu\poc\queue\stream\event\CallbackEvent;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup;
+use kuaukutsu\poc\queue\stream\internal\stream\RedisConsume;
 use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
 
 /**
@@ -40,7 +40,7 @@ final class Context
      */
     public function __construct(
         public readonly SchemaInterface $schema,
-        private readonly RedisStreamGroup $streamGroup,
+        private readonly RedisConsume $streamGroup,
         private readonly RedisString $string,
         private readonly EventDispatcher $eventDispatcher,
         public readonly int $maxExceededAttempts = 3,

src/internal/stream/RedisStreamGroup.php → src/internal/stream/RedisConsume.php

@@ -16,7 +16,7 @@
  * @psalm-internal kuaukutsu\poc\queue\stream
  * @psalm-suppress MissingThrowsDocblock
  */
-final readonly class RedisStreamGroup
+final readonly class RedisConsume
 {
     use ForbidCloning;
     use ForbidSerialization;
@@ -200,6 +200,9 @@ public function ack(string $identity, string ...$identities): bool
         return $result > 0;
     }
 
+    /**
+     * @see https://redis.io/docs/latest/commands/xgroup-delconsumer/
+     */
     public function delConsumer(): bool
     {
         $result = $this->client->execute(

src/internal/stream/RedisStream.php → src/internal/stream/RedisPublish.php

@@ -13,14 +13,22 @@
  * @psalm-internal kuaukutsu\poc\queue\stream
  * @psalm-suppress MissingThrowsDocblock
  */
-final readonly class RedisStream
+final readonly class RedisPublish
 {
     use ForbidCloning;
     use ForbidSerialization;
     use StreamUtils;
 
-    public function __construct(private RedisClient $client)
-    {
+    /**
+     * @var non-empty-string
+     */
+    private string $key;
+
+    public function __construct(
+        private RedisClient $client,
+        private SchemaInterface $schema,
+    ) {
+        $this->key = $this->generateKey($this->schema);
     }
 
     /**
@@ -29,12 +37,12 @@ public function __construct(private RedisClient $client)
      * @return ?non-empty-string
      * @see https://redis.io/docs/latest/commands/xadd/
      */
-    public function add(SchemaInterface $schema, array $payload, int|false $maxlen = 100_000): ?string
+    public function add(array $payload, int|false $maxlen = 100_000): ?string
     {
         $identity = $maxlen > 0
             ? $this->client->execute(
                 'XADD',
-                $this->generateKey($schema),
+                $this->key,
                 'MAXLEN',
                 '~',
                 $maxlen,
@@ -43,7 +51,7 @@ public function add(SchemaInterface $schema, array $payload, int|false $maxlen =
             )
             : $this->client->execute(
                 'XADD',
-                $this->generateKey($schema),
+                $this->key,
                 '*',
                 ...$this->preparePayload($payload)
             );

src/internal/stream/RedisString.php

@@ -7,14 +7,17 @@
 use Throwable;
 use Amp\Redis\Command\Option\SetOptions;
 use Amp\Redis\RedisClient;
+use kuaukutsu\queue\core\SchemaInterface;
 
 /**
  * @psalm-internal kuaukutsu\poc\queue\stream
  */
 final readonly class RedisString
 {
-    public function __construct(private RedisClient $client)
-    {
+    public function __construct(
+        private RedisClient $client,
+        private SchemaInterface $schema,
+    ) {
     }
 
     /**
@@ -25,7 +28,7 @@ public function __construct(private RedisClient $client)
     public function set(string $uuid, string $value, int $ttl = 600): bool
     {
         $options = new SetOptions();
-        return $this->client->set($uuid, $value, $options->withTtl($ttl));
+        return $this->client->set($this->generateKey($uuid), $value, $options->withTtl($ttl));
     }
 
     /**
@@ -35,8 +38,10 @@ public function set(string $uuid, string $value, int $ttl = 600): bool
      */
     public function copy(string $source, string $destination, int $ttl = 600): bool
     {
-        $row = $this->client->execute('COPY', $source, $destination);
-        if ($row > 0) {
+        $source = $this->generateKey($source);
+        $destination = $this->generateKey($destination);
+
+        if ($this->client->execute('COPY', $source, $destination) > 0) {
             return $this->client->expireIn($destination, $ttl);
         }
 
@@ -48,7 +53,7 @@ public function copy(string $source, string $destination, int $ttl = 600): bool
      */
     public function get(string $uuid): ?string
     {
-        $message = $this->client->get($uuid);
+        $message = $this->client->get($this->generateKey($uuid));
         if ($message === null || $message === '') {
             return null;
         }
@@ -63,9 +68,21 @@ public function get(string $uuid): ?string
     public function del(string $uuid, string ...$uuids): int
     {
         try {
-            return $this->client->delete($uuid, ...$uuids);
+            return $this->client->delete(
+                $this->generateKey($uuid),
+                ...array_map($this->generateKey(...), $uuids)
+            );
         } catch (Throwable) {
             return 0;
         }
     }
+
+    /**
+     * @param non-empty-string $uuid
+     * @return non-empty-string
+     */
+    private function generateKey(string $uuid): string
+    {
+        return hash('xxh3', $uuid . $this->schema->getRoutingKey());
+    }
 }

src/internal/workflow/WorkflowCatch.php

@@ -11,7 +11,7 @@
 use kuaukutsu\poc\queue\stream\event\MessageErrorEvent;
 use kuaukutsu\poc\queue\stream\event\SystemExceptionEvent;
 use kuaukutsu\poc\queue\stream\exception\WorkflowException;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup;
+use kuaukutsu\poc\queue\stream\internal\stream\RedisConsume;
 use kuaukutsu\poc\queue\stream\internal\Context;
 use kuaukutsu\poc\queue\stream\internal\Payload;
 
@@ -22,7 +22,7 @@
  */
 final readonly class WorkflowCatch
 {
-    public function __construct(private RedisStreamGroup $stream)
+    public function __construct(private RedisConsume $stream)
     {
     }
 
@@ -150,13 +150,13 @@ private function incrAttempt(Context $ctx, Payload $payload, int $currentAttempt
      * @param non-empty-string $identity
      * @return positive-int
      */
-    private function attempts(RedisStreamGroup $command, Context $ctx, string $identity): int
+    private function attempts(RedisConsume $command, Context $ctx, string $identity): int
     {
         /**
          * @param non-empty-string $identity
          * @return positive-int
          */
-        $fn = static function (RedisStreamGroup $command, Context $ctx, string $identity): int {
+        $fn = static function (RedisConsume $command, Context $ctx, string $identity): int {
             /** @var non-empty-string $identity */
             try {
                 $pending = $command->pending($identity);

src/internal/workflow/WorkflowClaim.php

@@ -8,7 +8,7 @@
 use Amp\TimeoutCancellation;
 use kuaukutsu\poc\queue\stream\event\Event;
 use kuaukutsu\poc\queue\stream\event\SystemExceptionEvent;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup;
+use kuaukutsu\poc\queue\stream\internal\stream\RedisConsume;
 use kuaukutsu\poc\queue\stream\internal\Context;
 use kuaukutsu\poc\queue\stream\internal\Payload;
 
@@ -22,7 +22,7 @@
 {
     public function __construct(
         private TaskHandler $action,
-        private RedisStreamGroup $stream,
+        private RedisConsume $stream,
     ) {
     }
 
@@ -64,12 +64,12 @@ public function __invoke(Context $ctx, WorkflowCatch $catch): void
     /**
      * @return iterable<non-empty-string, Payload>
      */
-    private function autoclaim(RedisStreamGroup $command, string $lastIdentity): iterable
+    private function autoclaim(RedisConsume $command, string $lastIdentity): iterable
     {
         /**
          * @return iterable<non-empty-string, Payload>
          */
-        $fn = static function (RedisStreamGroup $command, string $lastIdentity): iterable {
+        $fn = static function (RedisConsume $command, string $lastIdentity): iterable {
             $batch = $command->autoclaim($lastIdentity);
             if ($batch === []) {
                 return;

src/internal/workflow/WorkflowMain.php

@@ -8,7 +8,7 @@
 use Amp\TimeoutCancellation;
 use kuaukutsu\poc\queue\stream\event\Event;
 use kuaukutsu\poc\queue\stream\event\SystemExceptionEvent;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup;
+use kuaukutsu\poc\queue\stream\internal\stream\RedisConsume;
 use kuaukutsu\poc\queue\stream\internal\Context;
 use kuaukutsu\poc\queue\stream\internal\Payload;
 
@@ -22,7 +22,7 @@
 {
     public function __construct(
         private TaskHandler $action,
-        private RedisStreamGroup $stream,
+        private RedisConsume $stream,
     ) {
     }
 
@@ -65,12 +65,12 @@ public function __invoke(Context $ctx, WorkflowClaim $claim, WorkflowCatch $catc
     /**
      * @return iterable<non-empty-string, Payload>
      */
-    private function read(RedisStreamGroup $command): iterable
+    private function read(RedisConsume $command): iterable
     {
         /**
          * @return iterable<non-empty-string, Payload>
          */
-        $fn = static function (RedisStreamGroup $command): iterable {
+        $fn = static function (RedisConsume $command): iterable {
             $batch = $command->read();
             if ($batch === []) {
                 return;