From 08efa3979133ceacb0137eff40ec63b6583faa60 Mon Sep 17 00:00:00 2001 From: Dmitriy Krivopalov Date: Sun, 23 Nov 2025 16:45:32 +0300 Subject: [PATCH 1/2] event dispatcher --- .gitignore | 1 + benchmark/PublisherRedisBench.php | 34 ++++---- benchmark/PublisherValkeyBench.php | 34 ++++---- composer.json | 3 +- src/Builder.php | 26 +++++- src/Consumer.php | 15 +++- src/event/CallbackEvent.php | 20 +++++ src/event/Event.php | 24 +++++ src/event/EventDispatcher.php | 45 ++++++++++ src/event/EventInterface.php | 10 +++ src/event/EventPublisherInterface.php | 12 +++ src/event/EventSubscriberInterface.php | 13 +++ src/event/MessageAckEvent.php | 23 +++++ src/event/MessageErrorEvent.php | 30 +++++++ src/event/SystemExceptionEvent.php | 21 +++++ src/internal/Context.php | 27 +++++- src/internal/workflow/TaskHandler.php | 7 +- src/internal/workflow/WorkflowCatch.php | 77 ++++++++++------ src/internal/workflow/WorkflowClaim.php | 12 ++- src/internal/workflow/WorkflowMain.php | 41 ++++++--- src/tools/NullConsoleOutput.php | 111 ++++++++++++++++++++++++ src/tools/TraceConsoleOutput.php | 49 +++++++++++ tests/simulation/worker.php | 5 ++ 23 files changed, 549 insertions(+), 91 deletions(-) create mode 100644 src/event/CallbackEvent.php create mode 100644 src/event/Event.php create mode 100644 src/event/EventDispatcher.php create mode 100644 src/event/EventInterface.php create mode 100644 src/event/EventPublisherInterface.php create mode 100644 src/event/EventSubscriberInterface.php create mode 100644 src/event/MessageAckEvent.php create mode 100644 src/event/MessageErrorEvent.php create mode 100644 src/event/SystemExceptionEvent.php create mode 100644 src/tools/NullConsoleOutput.php create mode 100644 src/tools/TraceConsoleOutput.php diff --git a/.gitignore b/.gitignore index 7f64abe..e9e80d7 100644 --- a/.gitignore +++ b/.gitignore @@ -22,3 +22,4 @@ composer.lock phpunit.phar .phpunit.result.cache +*.local.php diff --git a/benchmark/PublisherRedisBench.php b/benchmark/PublisherRedisBench.php index a3e3a1e..4f51d47 100755 --- a/benchmark/PublisherRedisBench.php +++ b/benchmark/PublisherRedisBench.php @@ -42,18 +42,17 @@ public function benchAsWhile(): void // range foreach (range(1, 100) as $item) { - $this->publisher - ->push( - $schema, - new QueueTask( - target: QueueHandlerStub::class, - arguments: [ - 'id' => $item, - 'name' => 'bench range', - ], - ), - QueueContext::make($schema) - ); + $this->publisher->push( + $schema, + new QueueTask( + target: QueueHandlerStub::class, + arguments: [ + 'id' => $item, + 'name' => 'bench range', + ], + ), + QueueContext::make($schema) + ); } } @@ -72,11 +71,10 @@ public function benchAsBatch(): void ); } - $this->publisher - ->pushBatch( - $schema, - $batch, - QueueContext::make($schema) - ); + $this->publisher->pushBatch( + $schema, + $batch, + QueueContext::make($schema) + ); } } diff --git a/benchmark/PublisherValkeyBench.php b/benchmark/PublisherValkeyBench.php index 77c73f3..42c08c3 100755 --- a/benchmark/PublisherValkeyBench.php +++ b/benchmark/PublisherValkeyBench.php @@ -42,18 +42,17 @@ public function benchAsWhile(): void // range foreach (range(1, 100) as $item) { - $this->publisher - ->push( - $schema, - new QueueTask( - target: QueueHandlerStub::class, - arguments: [ - 'id' => $item, - 'name' => 'bench range', - ], - ), - QueueContext::make($schema) - ); + $this->publisher->push( + $schema, + new QueueTask( + target: QueueHandlerStub::class, + arguments: [ + 'id' => $item, + 'name' => 'bench range', + ], + ), + QueueContext::make($schema) + ); } } @@ -72,11 +71,10 @@ public function benchAsBatch(): void ); } - $this->publisher - ->pushBatch( - $schema, - $batch, - QueueContext::make($schema) - ); + $this->publisher->pushBatch( + $schema, + $batch, + QueueContext::make($schema) + ); } } diff --git a/composer.json b/composer.json index d3c8699..2b4a4c9 100644 --- a/composer.json +++ b/composer.json @@ -18,7 +18,8 @@ "php": "^8.3", "ext-redis": "*", "amphp/redis": "^2.0", - "kuaukutsu/queue-core": "^0.5.3" + "kuaukutsu/queue-core": "^0.5.3", + "symfony/console": "^7.3" }, "require-dev": { "ext-pcntl": "*", diff --git a/src/Builder.php b/src/Builder.php index f44264d..aad538d 100644 --- a/src/Builder.php +++ b/src/Builder.php @@ -6,6 +6,7 @@ use Closure; use Override; +use Throwable; use Amp\Redis\RedisConfig; use Amp\Redis\RedisException; use kuaukutsu\queue\core\handler\FactoryInterface; @@ -13,6 +14,8 @@ use kuaukutsu\queue\core\handler\Pipeline; use kuaukutsu\queue\core\interceptor\InterceptorInterface; use kuaukutsu\queue\core\BuilderInterface; +use kuaukutsu\poc\queue\stream\event\EventDispatcher; +use kuaukutsu\poc\queue\stream\event\EventSubscriberInterface; use function Amp\Redis\createRedisClient; @@ -27,6 +30,14 @@ final class Builder implements BuilderInterface private HandlerInterface $handler; + /** + * @var EventSubscriberInterface[] + */ + private array $eventSubscribers = []; + + /** + * @var ?Closure(?string, Throwable):void + */ private ?Closure $catch = null; /** @@ -59,6 +70,13 @@ public function withStreamOptions(StreamOptions $options): self return $clone; } + public function withSubscribers(EventSubscriberInterface ...$subscribers): self + { + $clone = clone $this; + $clone->eventSubscribers = $subscribers; + return $clone; + } + #[Override] public function withCatch(Closure $catch): BuilderInterface { @@ -84,6 +102,12 @@ public function buildPublisher(): Publisher #[Override] public function buildConsumer(): Consumer { - return new Consumer(createRedisClient($this->config), $this->options, $this->handler, $this->catch); + return new Consumer( + createRedisClient($this->config), + $this->options, + new EventDispatcher($this->eventSubscribers), + $this->handler, + $this->catch, + ); } } diff --git a/src/Consumer.php b/src/Consumer.php index 5a7008d..830ede9 100644 --- a/src/Consumer.php +++ b/src/Consumer.php @@ -6,19 +6,21 @@ 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\event\EventDispatcher; use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup; use kuaukutsu\poc\queue\stream\internal\stream\RedisString; 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; +use kuaukutsu\poc\queue\stream\internal\Context; /** * @api @@ -29,9 +31,13 @@ final class Consumer implements ConsumerInterface private readonly TaskHandler $handler; + /** + * @param ?Closure(?string, Throwable):void $catch + */ public function __construct( private readonly RedisClient $redis, private readonly StreamOptions $options, + private readonly EventDispatcher $eventDispatcher, HandlerInterface $handler, ?Closure $catch = null, ) { @@ -47,7 +53,12 @@ public function consume(SchemaInterface $schema): void $stream = new RedisStreamGroup($this->redis, $this->options, $schema); $stream->create(); - $this->ctx = new Context($schema, $stream, new RedisString($this->redis)); + $this->ctx = new Context( + $schema, + $stream, + new RedisString($this->redis), + $this->eventDispatcher, + ); EventLoop::queue( (new WorkflowMain($this->handler, $stream))(...), diff --git a/src/event/CallbackEvent.php b/src/event/CallbackEvent.php new file mode 100644 index 0000000..15e6bbe --- /dev/null +++ b/src/event/CallbackEvent.php @@ -0,0 +1,20 @@ +id); + } +} diff --git a/src/event/Event.php b/src/event/Event.php new file mode 100644 index 0000000..de40ebd --- /dev/null +++ b/src/event/Event.php @@ -0,0 +1,24 @@ +> + */ + private array $eventHandlers; + + /** + * @param EventSubscriberInterface[] $eventSubscribers + */ + public function __construct(array $eventSubscribers) + { + $subscriptions = []; + foreach ($eventSubscribers as $subscriber) { + foreach ($subscriber->subscriptions() as $name => $callback) { + $subscriptions[$name][] = $callback; + } + } + + $this->eventHandlers = $subscriptions; + } + + public function trigger(Event $name, EventInterface $event): void + { + if (array_key_exists($name->name, $this->eventHandlers)) { + foreach ($this->eventHandlers[$name->name] as $subscriberCallback) { + try { + $subscriberCallback($name, $event); + } catch (Throwable) { + } + } + } + } +} diff --git a/src/event/EventInterface.php b/src/event/EventInterface.php new file mode 100644 index 0000000..3c56fea --- /dev/null +++ b/src/event/EventInterface.php @@ -0,0 +1,10 @@ + + */ + public function subscriptions(): array; +} diff --git a/src/event/MessageAckEvent.php b/src/event/MessageAckEvent.php new file mode 100644 index 0000000..239c664 --- /dev/null +++ b/src/event/MessageAckEvent.php @@ -0,0 +1,23 @@ +identityList); + } +} diff --git a/src/event/MessageErrorEvent.php b/src/event/MessageErrorEvent.php new file mode 100644 index 0000000..e9da347 --- /dev/null +++ b/src/event/MessageErrorEvent.php @@ -0,0 +1,30 @@ +payload->uuid, + $this->attempt, + $this->exception->getMessage(), + ); + } +} diff --git a/src/event/SystemExceptionEvent.php b/src/event/SystemExceptionEvent.php new file mode 100644 index 0000000..834d418 --- /dev/null +++ b/src/event/SystemExceptionEvent.php @@ -0,0 +1,21 @@ +exception->getMessage()); + } +} diff --git a/src/internal/Context.php b/src/internal/Context.php index c85d1d2..e7e7504 100644 --- a/src/internal/Context.php +++ b/src/internal/Context.php @@ -7,6 +7,11 @@ use Closure; use Revolt\EventLoop; use kuaukutsu\queue\core\SchemaInterface; +use kuaukutsu\poc\queue\stream\event\Event; +use kuaukutsu\poc\queue\stream\event\EventInterface; +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\RedisString; @@ -37,6 +42,7 @@ public function __construct( public readonly SchemaInterface $schema, private readonly RedisStreamGroup $streamGroup, private readonly RedisString $string, + private readonly EventDispatcher $eventDispatcher, public readonly int $maxExceededAttempts = 3, ) { } @@ -69,6 +75,16 @@ public function copyData(string $source, string $destination, int $ttl = 600): b return $this->string->copy($source, $destination, $ttl); } + public function trigger(Event $name, EventInterface $event): void + { + $fn = $this->eventDispatcher->trigger(...); + EventLoop::defer( + static function () use ($fn, $name, $event): void { + $fn($name, $event); + } + ); + } + /** * @param non-empty-string $identity * @param non-empty-string $payloadUuid @@ -82,12 +98,13 @@ public function setAck(string $identity, string $payloadUuid): void public function sendAck(): void { if ($this->identityList !== []) { - $this->streamGroup->ack(array_shift($this->identityList), ...$this->identityList); + $this->streamGroup->ack($this->identityList[0], ...array_slice($this->identityList, 1)); + $this->trigger(Event::MessageAck, new MessageAckEvent($this->identityList)); $this->identityList = []; } if ($this->payloadList !== []) { - $this->string->del(array_shift($this->payloadList), ...$this->payloadList); + $this->string->del($this->payloadList[0], ...array_slice($this->payloadList, 1)); $this->payloadList = []; } } @@ -98,15 +115,17 @@ public function sendAck(): void public function defer(Closure $callback): void { $callbackId = EventLoop::defer($callback); + $this->trigger(Event::CallbackDeferred, new CallbackEvent($callbackId)); $this->callbackList[$callbackId] = true; // truncation tail - if (count($this->callbackList) > 32) { - $this->callbackList = array_slice($this->callbackList, -32, null, true); + if (count($this->callbackList) > 128) { + $this->callbackList = array_slice($this->callbackList, -128, null, true); } } public function done(string $callbackId): void { + $this->trigger(Event::CallbackDone, new CallbackEvent($callbackId)); unset($this->callbackList[$callbackId]); } diff --git a/src/internal/workflow/TaskHandler.php b/src/internal/workflow/TaskHandler.php index ae95665..af5459a 100644 --- a/src/internal/workflow/TaskHandler.php +++ b/src/internal/workflow/TaskHandler.php @@ -23,6 +23,9 @@ */ final readonly class TaskHandler { + /** + * @param ?Closure(?string, Throwable):void $catch + */ public function __construct( private HandlerInterface $handler, private ?Closure $catch, @@ -80,7 +83,7 @@ public function run(Closure $catchHandle, Context $context, string $identity, Pa try { async( $this->handler->handle(...), - $queueMessage + $queueMessage, )->await($cancellation); } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException $exception) { if ($this->tryCatch($message, $exception)) { @@ -118,7 +121,7 @@ public function run(Closure $catchHandle, Context $context, string $identity, Pa private function tryCatch(?string $message, Throwable $throwable): bool { if (is_callable($this->catch)) { - call_user_func($this->catch, $message, $throwable); + async($this->catch, $message, $throwable); return true; } diff --git a/src/internal/workflow/WorkflowCatch.php b/src/internal/workflow/WorkflowCatch.php index 6030534..f70e4cc 100644 --- a/src/internal/workflow/WorkflowCatch.php +++ b/src/internal/workflow/WorkflowCatch.php @@ -7,6 +7,9 @@ use Throwable; use Amp\CancelledException; use kuaukutsu\queue\core\QueueMessage; +use kuaukutsu\poc\queue\stream\event\Event; +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\Context; @@ -29,22 +32,41 @@ public function __construct(private RedisStreamGroup $stream) public function __invoke(Context $ctx, string $identity, Payload $payload, Throwable $exception): void { if ($exception instanceof WorkflowException) { + $ctx->trigger( + Event::MessageCorruptedError, + new MessageErrorEvent($payload, $exception), + ); + $this->dlq($ctx, $identity, $payload, $exception->getMessage()); return; } if ($exception instanceof CancelledException) { + $ctx->trigger( + Event::MessageTimeoutCancellation, + new MessageErrorEvent($payload, $exception), + ); + $this->dlq($ctx, $identity, $payload, $exception->getMessage()); return; } if ($ctx->maxExceededAttempts === 0) { + $ctx->trigger( + Event::MessageHandleError, + new MessageErrorEvent($payload, $exception), + ); + $this->dlq($ctx, $identity, $payload, $exception->getMessage()); return; } - $pending = $this->pending($this->stream, $identity); - $attempts = max(1, $pending['deliveryCount'] ?? 1); + $attempts = $this->attempts($this->stream, $ctx, $identity); + $ctx->trigger( + Event::MessageHandleError, + new MessageErrorEvent($payload, $exception, $attempts), + ); + if ($attempts >= $ctx->maxExceededAttempts) { $this->dlq( $ctx, @@ -76,7 +98,11 @@ private function dlq(Context $ctx, string $identity, Payload $payload, string $r 'target' => $payload->target, ], ); - } catch (Throwable) { + } catch (Throwable $exception) { + $ctx->trigger( + Event::RuntimeException, + new SystemExceptionEvent($exception), + ); } } @@ -112,45 +138,42 @@ private function incrAttempt(Context $ctx, Payload $payload, int $currentAttempt $queueMessage->context->incrAttempt(++$currentAttempt), ) ); - return; - } catch (Throwable) { - return; + } catch (Throwable $exception) { + $ctx->trigger( + Event::RuntimeException, + new SystemExceptionEvent($exception), + ); } } /** - * @return array{}|array{ - * "consumer": string, - * "elapsedMilliseconds": int, - * "deliveryCount": int, - * } + * @param non-empty-string $identity + * @return positive-int */ - private function pending(RedisStreamGroup $command, string $identity): array + private function attempts(RedisStreamGroup $command, Context $ctx, string $identity): int { /** * @param non-empty-string $identity - * @return array{}|array{ - * "consumer": string, - * "elapsedMilliseconds": int, - * "deliveryCount": int, - * } + * @return positive-int */ - $fn = static function (RedisStreamGroup $command, string $identity): array { + $fn = static function (RedisStreamGroup $command, Context $ctx, string $identity): int { /** @var non-empty-string $identity */ try { - return $command->pending($identity); - } catch (Throwable) { - return []; + $pending = $command->pending($identity); + } catch (Throwable $exception) { + $ctx->trigger( + Event::RuntimeException, + new SystemExceptionEvent($exception), + ); + return 1; } + + return max(1, $pending['deliveryCount']); }; /** - * @phpstan-var array{}|array{ - * "consumer": string, - * "elapsedMilliseconds": int, - * "deliveryCount": int, - * } + * @phpstan-var positive-int */ - return async($fn(...), $command, $identity)->await(); + return async($fn(...), $command, $ctx, $identity)->await(); } } diff --git a/src/internal/workflow/WorkflowClaim.php b/src/internal/workflow/WorkflowClaim.php index acc09f3..a685d5a 100644 --- a/src/internal/workflow/WorkflowClaim.php +++ b/src/internal/workflow/WorkflowClaim.php @@ -6,9 +6,11 @@ use Amp\CancelledException; 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\Context; use kuaukutsu\poc\queue\stream\internal\Payload; -use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup; use function Amp\async; use function Amp\Future\await; @@ -41,14 +43,16 @@ public function __invoke(Context $ctx, WorkflowCatch $catch): void } if ($list === []) { - $ctx->sendAck(); break; } try { await($list, new TimeoutCancellation(1800)); - } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException) { - // @fixme: logger + } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException $exception) { + $ctx->trigger( + Event::TimeoutCancellation, + new SystemExceptionEvent($exception), + ); } $ctx->sendAck(); diff --git a/src/internal/workflow/WorkflowMain.php b/src/internal/workflow/WorkflowMain.php index 5a06858..879183f 100644 --- a/src/internal/workflow/WorkflowMain.php +++ b/src/internal/workflow/WorkflowMain.php @@ -6,9 +6,11 @@ use Amp\CancelledException; 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\Context; use kuaukutsu\poc\queue\stream\internal\Payload; -use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup; use function Amp\async; use function Amp\Future\await; @@ -42,24 +44,21 @@ public function __invoke(Context $ctx, WorkflowClaim $claim, WorkflowCatch $catc $list[] = async($workflow(...), $identity, $payload); } + if ($list === []) { + $this->autoclaim($ctx, $claim, $catch, $lastAction); + continue; + } + try { await($list, new TimeoutCancellation(1800)); - } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException) { - // @fixme: logger + } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException $exception) { + $ctx->trigger( + Event::TimeoutCancellation, + new SystemExceptionEvent($exception), + ); } $ctx->sendAck(); - - if ($list === [] && $lastAction < strtotime('-30 seconds')) { - $ctx->defer( - static function (string $callbackId) use ($claim, $ctx, $catch): void { - $claim($ctx, $catch); - $ctx->done($callbackId); - } - ); - - $lastAction = time(); - } } } @@ -93,4 +92,18 @@ private function read(RedisStreamGroup $command): iterable */ return async($fn(...), $command)->await(); } + + private function autoclaim(Context $ctx, WorkflowClaim $claim, WorkflowCatch $catch, int &$lastAction): void + { + if ($lastAction < strtotime('-30 seconds')) { + $ctx->defer( + static function (string $callbackId) use ($claim, $ctx, $catch): void { + $claim($ctx, $catch); + $ctx->done($callbackId); + } + ); + + $lastAction = time(); + } + } } diff --git a/src/tools/NullConsoleOutput.php b/src/tools/NullConsoleOutput.php new file mode 100644 index 0000000..a72902d --- /dev/null +++ b/src/tools/NullConsoleOutput.php @@ -0,0 +1,111 @@ +output; + } + + #[Override] + public function setErrorOutput(OutputInterface $error): void + { + // do nothing + } + + /** + * @throws LogicException + */ + #[Override] + public function section(): never + { + throw new LogicException('nullable'); + } + + #[Override] + public function write(iterable | string $messages, bool $newline = false, int $options = 0): void + { + $this->output->write($messages, $newline, $options); + } + + #[Override] + public function writeln(iterable | string $messages, int $options = 0): void + { + $this->output->writeln($messages, $options); + } + + #[Override] + public function setVerbosity(int $level): void + { + $this->output->setVerbosity($level); + } + + #[Override] + public function getVerbosity(): int + { + return $this->output->getVerbosity(); + } + + #[Override] + public function isQuiet(): bool + { + return $this->output->isQuiet(); + } + + #[Override] + public function isVerbose(): bool + { + return $this->output->isVerbose(); + } + + #[Override] + public function isVeryVerbose(): bool + { + return $this->output->isVeryVerbose(); + } + + #[Override] + public function isDebug(): bool + { + return $this->output->isDebug(); + } + + #[Override] + public function setDecorated(bool $decorated): void + { + $this->output->setDecorated($decorated); + } + + #[Override] + public function isDecorated(): bool + { + return $this->output->isDecorated(); + } + + #[Override] + public function setFormatter(OutputFormatterInterface $formatter): void + { + $this->output->setFormatter($formatter); + } + + #[Override] + public function getFormatter(): OutputFormatterInterface + { + return $this->output->getFormatter(); + } +} diff --git a/src/tools/TraceConsoleOutput.php b/src/tools/TraceConsoleOutput.php new file mode 100644 index 0000000..daa551f --- /dev/null +++ b/src/tools/TraceConsoleOutput.php @@ -0,0 +1,49 @@ +name] = $this->trace(...); + } + + /** + * @var non-empty-array $subscriptions + * @phpstan-ignore varTag.nativeType + */ + return $subscriptions; + } + + public function trace(Event $name, EventInterface $event): void + { + $this->stdout( + sprintf( + '[%s] %s', + $name->value, + $event->getMessage(), + ) + ); + } + + private function stdout(string $message): void + { + $this->output->writeln($message); + } +} diff --git a/tests/simulation/worker.php b/tests/simulation/worker.php index 82eea02..0cf0150 100644 --- a/tests/simulation/worker.php +++ b/tests/simulation/worker.php @@ -7,6 +7,8 @@ declare(strict_types=1); +use Symfony\Component\Console\Output\ConsoleOutput; +use kuaukutsu\poc\queue\stream\tools\TraceConsoleOutput; use kuaukutsu\queue\core\interceptor\ArgumentsVerifyInterceptor; use kuaukutsu\poc\queue\stream\interceptor\ExactlyOnceInterceptor; use kuaukutsu\poc\queue\stream\tests\stub\QueueSchemaStub; @@ -22,6 +24,9 @@ echo 'consumer run: ' . $schema->getRoutingKey() . PHP_EOL; $consumer = $builder + ->withSubscribers( + new TraceConsoleOutput(new ConsoleOutput()) + ) ->withInterceptors( new ArgumentsVerifyInterceptor(), new ExactlyOnceInterceptor(createRedisClient('tcp://redis:6379')), From 4eed661babe10a98c9151ba2bb0f48cf8a341af7 Mon Sep 17 00:00:00 2001 From: Dmitriy Krivopalov Date: Mon, 24 Nov 2025 08:43:09 +0300 Subject: [PATCH 2/2] fix simulation --- src/event/MessageAckEvent.php | 2 +- src/internal/workflow/WorkflowClaim.php | 12 ++++++---- tests/simulation/publisher-batch.php | 2 +- tests/simulation/publisher.php | 6 ++--- tests/stub/QueueExceptionHandlerStub.php | 30 ++++++++++++++++++++++++ tests/stub/QueueHandlerStub.php | 5 ---- 6 files changed, 42 insertions(+), 15 deletions(-) create mode 100644 tests/stub/QueueExceptionHandlerStub.php diff --git a/src/event/MessageAckEvent.php b/src/event/MessageAckEvent.php index 239c664..30f856c 100644 --- a/src/event/MessageAckEvent.php +++ b/src/event/MessageAckEvent.php @@ -18,6 +18,6 @@ public function __construct(public array $identityList) #[Override] public function getMessage(): string { - return implode(', ', $this->identityList); + return sprintf('COUNT: %d, %s', count($this->identityList), implode(', ', $this->identityList)); } } diff --git a/src/internal/workflow/WorkflowClaim.php b/src/internal/workflow/WorkflowClaim.php index a685d5a..58a633b 100644 --- a/src/internal/workflow/WorkflowClaim.php +++ b/src/internal/workflow/WorkflowClaim.php @@ -36,10 +36,12 @@ public function __invoke(Context $ctx, WorkflowCatch $catch): void } }; + $lastIdentity = '0-0'; while (true) { $list = []; - foreach ($this->autoclaim($this->stream) as $identity => $payload) { + foreach ($this->autoclaim($this->stream, $lastIdentity) as $identity => $payload) { $list[] = async($workflow(...), $identity, $payload); + $lastIdentity = $identity; } if ($list === []) { @@ -62,13 +64,13 @@ public function __invoke(Context $ctx, WorkflowCatch $catch): void /** * @return iterable */ - private function autoclaim(RedisStreamGroup $command): iterable + private function autoclaim(RedisStreamGroup $command, string $lastIdentity): iterable { /** * @return iterable */ - $fn = static function (RedisStreamGroup $command): iterable { - $batch = $command->autoclaim(); + $fn = static function (RedisStreamGroup $command, string $lastIdentity): iterable { + $batch = $command->autoclaim($lastIdentity); if ($batch === []) { return; } @@ -87,6 +89,6 @@ private function autoclaim(RedisStreamGroup $command): iterable /** * @phpstan-var iterable */ - return async($fn(...), $command)->await(); + return async($fn(...), $command, $lastIdentity)->await(); } } diff --git a/tests/simulation/publisher-batch.php b/tests/simulation/publisher-batch.php index e102502..9f8c9a8 100644 --- a/tests/simulation/publisher-batch.php +++ b/tests/simulation/publisher-batch.php @@ -21,7 +21,7 @@ echo 'publisher run: ' . $schema->getRoutingKey() . PHP_EOL; $batch = []; -foreach (range(1, 100) as $item) { +foreach (range(1, 500) as $item) { $batch[] = new QueueTask( target: QueueHandlerStub::class, arguments: [ diff --git a/tests/simulation/publisher.php b/tests/simulation/publisher.php index 790ca4d..73835de 100644 --- a/tests/simulation/publisher.php +++ b/tests/simulation/publisher.php @@ -8,7 +8,7 @@ declare(strict_types=1); use kuaukutsu\poc\queue\stream\Builder; -use kuaukutsu\poc\queue\stream\tests\stub\QueueHandlerStub; +use kuaukutsu\poc\queue\stream\tests\stub\QueueExceptionHandlerStub; use kuaukutsu\poc\queue\stream\tests\stub\QueueSchemaStub; use kuaukutsu\queue\core\QueueContext; use kuaukutsu\queue\core\QueueTask; @@ -23,7 +23,7 @@ $publisher = $builder->buildPublisher(); $task = new QueueTask( - target: QueueHandlerStub::class, + target: QueueExceptionHandlerStub::class, arguments: [ 'id' => 1, 'name' => 'test name', @@ -37,7 +37,7 @@ $publisher->push( $schema, new QueueTask( - target: QueueHandlerStub::class, + target: QueueExceptionHandlerStub::class, arguments: [ 'id' => $item, 'name' => 'test range', diff --git a/tests/stub/QueueExceptionHandlerStub.php b/tests/stub/QueueExceptionHandlerStub.php new file mode 100644 index 0000000..93574c8 --- /dev/null +++ b/tests/stub/QueueExceptionHandlerStub.php @@ -0,0 +1,30 @@ +attempt === 1 && random_int(0, 1) === 1) { + throw new LogicException('Random exception.'); + } + + $this->writer->print($this->id, $this->name, $context); + } +} diff --git a/tests/stub/QueueHandlerStub.php b/tests/stub/QueueHandlerStub.php index 6cfc315..1dd2508 100644 --- a/tests/stub/QueueHandlerStub.php +++ b/tests/stub/QueueHandlerStub.php @@ -5,7 +5,6 @@ namespace kuaukutsu\poc\queue\stream\tests\stub; use Override; -use LogicException; use kuaukutsu\queue\core\QueueContext; use kuaukutsu\queue\core\TaskInterface; @@ -21,10 +20,6 @@ public function __construct( #[Override] public function handle(QueueContext $context): void { - if ($context->attempt === 1 && random_int(0, 1) === 1) { - throw new LogicException('Random exception.'); - } - $this->writer->print($this->id, $this->name, $context); } }