diff --git a/.gitattributes b/.gitattributes index 1acef8c..33245e5 100644 --- a/.gitattributes +++ b/.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 diff --git a/src/Consumer.php b/src/Consumer.php index 830ede9..c3abf39 100644 --- a/src/Consumer.php +++ b/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, ); diff --git a/src/Publisher.php b/src/Publisher.php index 1afe852..84d56a9 100644 --- a/src/Publisher.php +++ b/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); } diff --git a/src/internal/Context.php b/src/internal/Context.php index e7e7504..3e73f9c 100644 --- a/src/internal/Context.php +++ b/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, diff --git a/src/internal/stream/RedisStreamGroup.php b/src/internal/stream/RedisConsume.php similarity index 97% rename from src/internal/stream/RedisStreamGroup.php rename to src/internal/stream/RedisConsume.php index 2e40b7c..c0b6958 100644 --- a/src/internal/stream/RedisStreamGroup.php +++ b/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( diff --git a/src/internal/stream/RedisStream.php b/src/internal/stream/RedisPublish.php similarity index 73% rename from src/internal/stream/RedisStream.php rename to src/internal/stream/RedisPublish.php index 3fc9ee8..18de668 100644 --- a/src/internal/stream/RedisStream.php +++ b/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) ); diff --git a/src/internal/stream/RedisString.php b/src/internal/stream/RedisString.php index c490b03..4b2dca8 100644 --- a/src/internal/stream/RedisString.php +++ b/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()); + } } diff --git a/src/internal/workflow/WorkflowCatch.php b/src/internal/workflow/WorkflowCatch.php index f70e4cc..98c4a26 100644 --- a/src/internal/workflow/WorkflowCatch.php +++ b/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); diff --git a/src/internal/workflow/WorkflowClaim.php b/src/internal/workflow/WorkflowClaim.php index 58a633b..5ccfbd7 100644 --- a/src/internal/workflow/WorkflowClaim.php +++ b/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 */ - private function autoclaim(RedisStreamGroup $command, string $lastIdentity): iterable + private function autoclaim(RedisConsume $command, string $lastIdentity): iterable { /** * @return iterable */ - $fn = static function (RedisStreamGroup $command, string $lastIdentity): iterable { + $fn = static function (RedisConsume $command, string $lastIdentity): iterable { $batch = $command->autoclaim($lastIdentity); if ($batch === []) { return; diff --git a/src/internal/workflow/WorkflowMain.php b/src/internal/workflow/WorkflowMain.php index 879183f..3b46c07 100644 --- a/src/internal/workflow/WorkflowMain.php +++ b/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 */ - private function read(RedisStreamGroup $command): iterable + private function read(RedisConsume $command): iterable { /** * @return iterable */ - $fn = static function (RedisStreamGroup $command): iterable { + $fn = static function (RedisConsume $command): iterable { $batch = $command->read(); if ($batch === []) { return;