consumer refactoring

Makefile

@@ -84,7 +84,7 @@ publisher:
 
 consumer:
 	USER=$(USER) docker compose -f ./docker-compose.yml run --rm -u $(USER) -w /tests/simulation cli \
-		php worker.php --schema=high
+		php worker-with-catch.php --schema=high
 
 bench: ## bench
 	USER=$(USER) docker compose -f ./docker-compose.yml run --rm -u $(USER) -w / cli \

README.md

@@ -51,6 +51,82 @@ composer require kuaukutsu/poc-queue-stream
 - https://redis.io/docs/latest/develop/data-types/streams/
 - https://habr.com/ru/articles/456270/
 
+### example
+
+`tests/simulation`
+
+#### common
+
+```php
+// необходим для внедрения зависимостей в объект задачи
+$container = new Container();
+$builder = (new Builder(new FactoryProxy($container)))
+    ->withConfig(
+        RedisConfig::fromUri('tcp://redis:6379')
+    );
+
+// например из аргументов получаем схему - название очереди
+$schema = QueueSchemaStub::from((string)argument('schema', 'low'));
+```
+
+#### publish
+
+```php
+// где-то в своём проекте (app), через Interface и Decorator заводим сервис PublisherInterface
+$publisher = $builder->buildPublisher();
+
+// создание отложенной задачи
+$task = new QueueTask(
+    target: QueueHandlerStub::class,
+    arguments: [
+        'id' => 1,
+        'name' => 'test name',
+    ],
+);
+
+$publisher->push($schema, $task);
+```
+
+Или массовая публикация
+```php
+$batch = [];
+foreach (range(1, 100) as $item) {
+    $batch[] = new QueueTask(
+        target: QueueHandlerStub::class,
+        arguments: [
+            'id' => $item,
+            'name' => 'test batch',
+        ],
+    );
+}
+
+$publisher->pushBatch($schema, $batch);
+```
+
+#### consume
+
+consumer.php
+```php
+$consumer = $builder
+    // задаём обработчик ошибок, если необходимо и не устраивает глобальный TryCatchInterceptor
+    ->withCatch(
+        static function (?string $message, Throwable $exception): void {
+            echo sprintf("data: %s\nerror: %s", $message, $exception->getMessage());
+        }
+    )
+    ->withInterceptors(
+        new ArgumentsVerifyInterceptor(),
+        new ExactlyOnceInterceptor(createRedisClient('redis://redis:6379')),
+        new TryCatchInterceptor(),
+    )
+    ->buildConsumer();
+
+$consumer->consume($schema);
+trapSignal([SIGTERM, SIGINT]);
+$consumer->disconnect();
+exit(0);
+```
+
 ### Benchmark
 
 ```
@@ -64,21 +140,21 @@ with PHP version 8.3.17, xdebug ✔, opcache ✔
 
 \kuaukutsu\poc\queue\stream\benchmarks\PublisherRedisBench
 
-    benchAsWhile............................I4 - Mo32.201ms (±5.32%)
-    benchAsBatch............................I4 - Mo29.561ms (±1.30%)
+    benchAsWhile............................I4 - Mo29.287ms (±15.55%)
+    benchAsBatch............................I4 - Mo17.785ms (±2.50%)
 
 \kuaukutsu\poc\queue\stream\benchmarks\PublisherValkeyBench
 
-    benchAsWhile............................I4 - Mo31.663ms (±2.65%)
-    benchAsBatch............................I4 - Mo30.349ms (±3.98%)
+    benchAsWhile............................I4 - Mo30.377ms (±8.72%)
+    benchAsBatch............................I4 - Mo17.267ms (±3.16%)
 
 Subjects: 4, Assertions: 0, Failures: 0, Errors: 0
-+----------------------+--------------+-----+------+-----+----------+----------+--------+
-| benchmark            | subject      | set | revs | its | mem_peak | mode     | rstdev |
-+----------------------+--------------+-----+------+-----+----------+----------+--------+
-| PublisherRedisBench  | benchAsWhile |     | 10   | 5   | 1.983mb  | 32.201ms | ±5.32% |
-| PublisherRedisBench  | benchAsBatch |     | 10   | 5   | 2.044mb  | 29.561ms | ±1.30% |
-| PublisherValkeyBench | benchAsWhile |     | 10   | 5   | 1.983mb  | 31.663ms | ±2.65% |
-| PublisherValkeyBench | benchAsBatch |     | 10   | 5   | 2.037mb  | 30.349ms | ±3.98% |
-+----------------------+--------------+-----+------+-----+----------+----------+--------+
++----------------------+--------------+-----+------+-----+----------+----------+---------+
+| benchmark            | subject      | set | revs | its | mem_peak | mode     | rstdev  |
++----------------------+--------------+-----+------+-----+----------+----------+---------+
+| PublisherRedisBench  | benchAsWhile |     | 10   | 5   | 2.053mb  | 29.287ms | ±15.55% |
+| PublisherRedisBench  | benchAsBatch |     | 10   | 5   | 4.589mb  | 17.785ms | ±2.50%  |
+| PublisherValkeyBench | benchAsWhile |     | 10   | 5   | 2.046mb  | 30.377ms | ±8.72%  |
+| PublisherValkeyBench | benchAsBatch |     | 10   | 5   | 4.589mb  | 17.267ms | ±3.16%  |
++----------------------+--------------+-----+------+-----+----------+----------+---------+
 ```

composer.json

@@ -18,7 +18,7 @@
     "php": "^8.3",
     "ext-redis": "*",
     "amphp/redis": "^2.0",
-    "kuaukutsu/queue-core": "^0.4"
+    "kuaukutsu/queue-core": "^0.5.2"
   },
   "require-dev": {
     "ext-pcntl": "*",

phpstan.neon

@@ -6,4 +6,3 @@ parameters:
 	ignoreErrors:
 		- identifiers:
 		  - missingType.iterableValue
-		  - missingType.generics

src/Builder.php

@@ -4,6 +4,7 @@
 
 namespace kuaukutsu\poc\queue\stream;
 
+use Closure;
 use Override;
 use Amp\Redis\RedisConfig;
 use Amp\Redis\RedisException;
@@ -12,10 +13,6 @@
 use kuaukutsu\queue\core\handler\Pipeline;
 use kuaukutsu\queue\core\interceptor\InterceptorInterface;
 use kuaukutsu\queue\core\BuilderInterface;
-use kuaukutsu\queue\core\SchemaInterface;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStream;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
 
 use function Amp\Redis\createRedisClient;
 
@@ -30,6 +27,8 @@ final class Builder implements BuilderInterface
 
     private HandlerInterface $handler;
 
+    private ?Closure $catch = null;
+
     /**
      * @throws RedisException
      */
@@ -60,6 +59,14 @@ public function withStreamOptions(StreamOptions $options): self
         return $clone;
     }
 
+    #[Override]
+    public function withCatch(Closure $catch): BuilderInterface
+    {
+        $clone = clone $this;
+        $clone->catch = $catch;
+        return $clone;
+    }
+
     #[Override]
     public function withInterceptors(InterceptorInterface ...$interceptor): self
     {
@@ -75,15 +82,8 @@ public function buildPublisher(): Publisher
     }
 
     #[Override]
-    public function buildConsumer(SchemaInterface $schema): Consumer
+    public function buildConsumer(): Consumer
     {
-        $client = createRedisClient($this->config);
-
-        return new Consumer(
-            new RedisStream($client, $schema),
-            new RedisStreamGroup($client, $schema, $this->options),
-            new RedisString($client),
-            $this->handler,
-        );
+        return new Consumer(createRedisClient($this->config), $this->options, $this->handler, $this->catch);
     }
 }

src/Consumer.php

@@ -4,66 +4,64 @@
 
 namespace kuaukutsu\poc\queue\stream;
 
+use Closure;
 use Override;
-use Throwable;
+use Amp\Redis\RedisClient;
 use Revolt\EventLoop;
 use kuaukutsu\queue\core\exception\QueueConsumeException;
 use kuaukutsu\queue\core\handler\HandlerInterface;
 use kuaukutsu\queue\core\ConsumerInterface;
+use kuaukutsu\queue\core\SchemaInterface;
 use kuaukutsu\poc\queue\stream\internal\Context;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStream;
 use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup;
 use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
-use kuaukutsu\poc\queue\stream\internal\workflow\TaskRunner;
+use kuaukutsu\poc\queue\stream\internal\workflow\TaskHandler;
+use kuaukutsu\poc\queue\stream\internal\workflow\WorkflowCatch;
 use kuaukutsu\poc\queue\stream\internal\workflow\WorkflowClaim;
 use kuaukutsu\poc\queue\stream\internal\workflow\WorkflowMain;
 
 /**
  * @api
  */
-final readonly class Consumer implements ConsumerInterface
+final class Consumer implements ConsumerInterface
 {
-    private Context $ctx;
+    private ?Context $ctx = null;
 
-    private TaskRunner $runner;
+    private readonly TaskHandler $handler;
 
     public function __construct(
-        RedisStream $stream,
-        private RedisStreamGroup $streamGroup,
-        RedisString $string,
+        private readonly RedisClient $redis,
+        private readonly StreamOptions $options,
         HandlerInterface $handler,
+        ?Closure $catch = null,
     ) {
-        $this->ctx = new Context($string, $this->streamGroup);
-        $this->runner = new TaskRunner($string, $stream, $this->streamGroup, $handler);
+        $this->handler = new TaskHandler($handler, $catch);
     }
 
     /**
-     * @param ?callable(string, Throwable):void $catch
      * @throws QueueConsumeException
      */
     #[Override]
-    public function consume(?callable $catch = null): void
+    public function consume(SchemaInterface $schema): void
     {
-        $this->streamGroup->create();
+        $stream = new RedisStreamGroup($this->redis, $this->options, $schema);
+        $stream->create();
 
-        $worflow = new WorkflowMain(
-            $this->runner,
-            $this->streamGroup,
-        );
+        $this->ctx = new Context($schema, $stream, new RedisString($this->redis));
 
         EventLoop::queue(
-            $worflow(...),
-            $this->ctx->withCatch($catch),
-            new WorkflowClaim(
-                $this->runner,
-                $this->streamGroup,
-            ),
+            (new WorkflowMain($this->handler, $stream))(...),
+            $this->ctx,
+            new WorkflowClaim($this->handler, $stream),
+            new WorkflowCatch($stream),
         );
     }
 
+    #[Override]
     public function disconnect(): void
     {
-        $this->ctx->cancel();
-        $this->streamGroup->delConsumer();
+        if ($this->ctx instanceof Context) {
+            $this->ctx->cancel();
+        }
     }
 }

src/Publisher.php

@@ -6,28 +6,33 @@
 
 use Override;
 use Throwable;
-use WeakMap;
 use Amp\Redis\RedisClient;
-use kuaukutsu\poc\queue\stream\internal\Payload;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStream;
 use kuaukutsu\queue\core\exception\QueuePublishException;
 use kuaukutsu\queue\core\PublisherInterface;
 use kuaukutsu\queue\core\QueueContext;
 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\RedisString;
+use kuaukutsu\poc\queue\stream\internal\Payload;
+
+use function Amp\async;
+use function Amp\Future\await;
 
 /**
  * @api
  */
-final class Publisher implements PublisherInterface
+final readonly class Publisher implements PublisherInterface
 {
-    private WeakMap $map;
+    private RedisStream $stream;
 
-    public function __construct(private readonly RedisClient $client)
+    private RedisString $string;
+
+    public function __construct(RedisClient $redis)
     {
-        $this->map = new WeakMap();
+        $this->stream = new RedisStream($redis);
+        $this->string = new RedisString($redis);
     }
 
     /**
@@ -37,45 +42,32 @@ public function __construct(private readonly RedisClient $client)
     #[Override]
     public function push(SchemaInterface $schema, QueueTask $task, ?QueueContext $context = null): string
     {
-        $storage = new RedisString($this->client);
-        $storage->set(
+        $this->string->set(
             $task->getUuid(),
             QueueMessage::makeMessage($task, $context ?? QueueContext::make($schema)),
         );
 
         try {
-            $this->makeStream($schema)->add(
-                Payload::fromTask($task)->toArray()
-            );
+            $this->stream->add($schema, Payload::fromTask($task)->toArray());
         } catch (Throwable $exception) {
-            $storage->del($task->getUuid());
+            $this->string->del($task->getUuid());
             throw new QueuePublishException($schema, $exception);
         }
 
         return $task->getUuid();
     }
 
-    /**
-     * @param list<QueueTask> $taskBatch
-     * @return list<non-empty-string>
-     * @throws QueuePublishException
-     */
     #[Override]
     public function pushBatch(SchemaInterface $schema, array $taskBatch, ?QueueContext $context = null): array
     {
-        $list = [];
+        $groupAwait = [];
         foreach ($taskBatch as $task) {
-            $list[] = $this->push($schema, $task, $context);
+            $groupAwait[] = async($this->push(...), $schema, $task, $context);
         }
 
-        return $list;
-    }
-
-    private function makeStream(SchemaInterface $schema): RedisStream
-    {
         /**
-         * @var RedisStream
+         * @var list<non-empty-string>
          */
-        return $this->map[$schema] ??= new RedisStream($this->client, $schema);
+        return await($groupAwait);
     }
 }