diff --git a/src/Consumer.php b/src/Consumer.php index e536684..fd2d037 100644 --- a/src/Consumer.php +++ b/src/Consumer.php @@ -8,6 +8,7 @@ use Override; use Throwable; use Amp\Redis\RedisClient; +use Amp\TimeoutCancellation; use Revolt\EventLoop; use kuaukutsu\queue\core\handler\HandlerInterface; use kuaukutsu\queue\core\ConsumerInterface; @@ -55,6 +56,7 @@ public function consume(SchemaInterface $schema): void $stream, $string, $this->eventDispatcher, + new TimeoutCancellation(1800), ); EventLoop::queue( diff --git a/src/internal/Context.php b/src/internal/Context.php index 24ff350..89d45a8 100644 --- a/src/internal/Context.php +++ b/src/internal/Context.php @@ -5,6 +5,9 @@ namespace kuaukutsu\poc\queue\stream\internal; use Closure; +use Amp\Future; +use Amp\Cancellation; +use Amp\CancelledException; use Revolt\EventLoop; use kuaukutsu\queue\core\SchemaInterface; use kuaukutsu\poc\queue\stream\event\Event; @@ -12,9 +15,12 @@ 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\event\SystemExceptionEvent; use kuaukutsu\poc\queue\stream\internal\stream\RedisConsume; use kuaukutsu\poc\queue\stream\internal\stream\RedisString; +use function Amp\Future\await; + /** * @psalm-internal kuaukutsu\poc\queue\stream */ @@ -40,9 +46,20 @@ public function __construct( private readonly RedisConsume $streamGroup, private readonly RedisString $string, private readonly EventDispatcher $eventDispatcher, + private readonly ?Cancellation $cancellation = null, ) { } + 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 $uuid */ @@ -71,16 +88,6 @@ 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 @@ -105,6 +112,43 @@ public function sendAck(): void } } + /** + * @template T + * @param Future $future + * @return ?T + * @noinspection PhpRedundantCatchClauseInspection + */ + public function awaitFuture(Future $future): mixed + { + try { + return $future->await($this->cancellation); + } catch (CancelledException $exception) { + $this->trigger( + Event::TimeoutCancellation, + new SystemExceptionEvent($exception), + ); + } + + return null; + } + + /** + * @template T + * @param list> $futures + * @noinspection PhpRedundantCatchClauseInspection + */ + public function awaitFutures(array $futures): void + { + try { + await($futures, $this->cancellation); + } catch (CancelledException $exception) { + $this->trigger( + Event::TimeoutCancellation, + new SystemExceptionEvent($exception), + ); + } + } + /** * @param Closure(string):void $callback */ diff --git a/src/internal/workflow/TaskHandler.php b/src/internal/workflow/TaskHandler.php index af5459a..afe3184 100644 --- a/src/internal/workflow/TaskHandler.php +++ b/src/internal/workflow/TaskHandler.php @@ -33,11 +33,11 @@ public function __construct( } /** - * @param Closure(Context, non-empty-string, Payload, Throwable):void $catchHandle * @param non-empty-string $identity + * @param Closure(Context, non-empty-string, Payload, Throwable):void $catchHandle * @return bool TRUE отправить ACK; FALSE не отправлять ACK */ - public function run(Closure $catchHandle, Context $context, string $identity, Payload $payload): bool + public function run(Context $context, string $identity, Payload $payload, Closure $catchHandle): bool { $message = $context->getData($payload->uuid); if ($message === null || $message === '') { diff --git a/src/internal/workflow/WorkflowCatch.php b/src/internal/workflow/WorkflowCatch.php index ef686f9..89e813b 100644 --- a/src/internal/workflow/WorkflowCatch.php +++ b/src/internal/workflow/WorkflowCatch.php @@ -1,5 +1,7 @@ attempts($this->stream, $ctx, $identity); + $attempts = $this->attempts($ctx, $this->stream, $identity); $ctx->trigger( Event::MessageHandleError, new MessageErrorEvent($payload, $exception, $attempts), @@ -155,29 +157,29 @@ private function incrAttempt(Context $ctx, Payload $payload, int $currentAttempt * @param non-empty-string $identity * @return positive-int */ - private function attempts(RedisConsume $command, Context $ctx, string $identity): int + private function attempts(Context $ctx, RedisConsume $command, string $identity): int { /** - * @psalm-var Closure(RedisConsume, Context, non-empty-string): positive-int $fn + * @psalm-var Closure(Context, RedisConsume, non-empty-string): positive-int $fn */ - static $fn = static function (RedisConsume $command, Context $ctx, string $identity): int { + static $fn = static function (Context $ctx, RedisConsume $command, string $identity): int { /** @var non-empty-string $identity */ try { $pending = $command->pending($identity); + return max(1, $pending['deliveryCount']); } catch (Throwable $exception) { $ctx->trigger( Event::RuntimeException, new SystemExceptionEvent($exception), ); - return 1; } - return max(1, $pending['deliveryCount']); + return 1; }; /** * @phpstan-var positive-int */ - return async($fn(...), $command, $ctx, $identity)->await(); + return $ctx->awaitFuture(async($fn(...), $ctx, $command, $identity)) ?? 1; } } diff --git a/src/internal/workflow/WorkflowClaim.php b/src/internal/workflow/WorkflowClaim.php index 12a2632..1ba3430 100644 --- a/src/internal/workflow/WorkflowClaim.php +++ b/src/internal/workflow/WorkflowClaim.php @@ -1,20 +1,17 @@ action->run(...); + static $action = $this->action->run(...); /** * @psalm-var Closure(Context, string, Payload, callable, callable): void $workflow @@ -42,7 +39,7 @@ public function __invoke(Context $ctx, WorkflowCatch $catch): void callable $catch, ): void { /** @var non-empty-string $identity */ - if ($action($catch(...), $ctx, $identity, $payload)) { + if ($action($ctx, $identity, $payload, $catch(...))) { $ctx->setAck($identity, $payload->uuid); } }; @@ -50,7 +47,7 @@ public function __invoke(Context $ctx, WorkflowCatch $catch): void $lastIdentity = '0-0'; while (true) { $list = []; - foreach ($this->autoclaim($this->stream, $lastIdentity) as $identity => $payload) { + foreach ($this->autoclaim($ctx, $this->stream, $lastIdentity) as $identity => $payload) { $list[] = async($workflow(...), $ctx, $identity, $payload, $action, $catch); $lastIdentity = $identity; } @@ -59,15 +56,7 @@ public function __invoke(Context $ctx, WorkflowCatch $catch): void break; } - try { - await($list, new TimeoutCancellation(1800)); - } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException $exception) { - $ctx->trigger( - Event::TimeoutCancellation, - new SystemExceptionEvent($exception), - ); - } - + $ctx->awaitFutures($list); $ctx->sendAck(); } } @@ -75,7 +64,7 @@ public function __invoke(Context $ctx, WorkflowCatch $catch): void /** * @return iterable */ - private function autoclaim(RedisConsume $command, string $lastIdentity): iterable + private function autoclaim(Context $ctx, RedisConsume $command, string $lastIdentity): iterable { /** * @psalm-var Closure(RedisConsume, string): iterable $fn @@ -99,8 +88,8 @@ private function autoclaim(RedisConsume $command, string $lastIdentity): iterabl }; /** - * @phpstan-var iterable + * @var iterable */ - return async($fn(...), $command, $lastIdentity)->await(); + return $ctx->awaitFuture(async($fn(...), $command, $lastIdentity)) ?? []; } } diff --git a/src/internal/workflow/WorkflowMain.php b/src/internal/workflow/WorkflowMain.php index 1f64dad..7f07dd1 100644 --- a/src/internal/workflow/WorkflowMain.php +++ b/src/internal/workflow/WorkflowMain.php @@ -5,16 +5,11 @@ namespace kuaukutsu\poc\queue\stream\internal\workflow; use Closure; -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\RedisConsume; use kuaukutsu\poc\queue\stream\internal\Context; use kuaukutsu\poc\queue\stream\internal\Payload; use function Amp\async; -use function Amp\Future\await; /** * @psalm-internal kuaukutsu\poc\queue\stream @@ -29,8 +24,7 @@ public function __construct( public function __invoke(Context $ctx, WorkflowClaim $claim, WorkflowCatch $catch): void { - $lastAction = time(); - $action = $this->action->run(...); + static $action = $this->action->run(...); /** * @psalm-var Closure(Context, string, Payload, callable, callable): void $workflow @@ -43,15 +37,17 @@ public function __invoke(Context $ctx, WorkflowClaim $claim, WorkflowCatch $catc callable $catch, ): void { /** @var non-empty-string $identity */ - if ($action($catch(...), $ctx, $identity, $payload)) { + if ($action($ctx, $identity, $payload, $catch(...))) { $ctx->setAck($identity, $payload->uuid); } }; + $lastAction = time(); + /** @phpstan-ignore while.alwaysTrue */ while (true) { $list = []; - foreach ($this->read($this->stream) as $identity => $payload) { + foreach ($this->read($ctx, $this->stream) as $identity => $payload) { $list[] = async($workflow(...), $ctx, $identity, $payload, $action, $catch); } @@ -60,15 +56,7 @@ public function __invoke(Context $ctx, WorkflowClaim $claim, WorkflowCatch $catc continue; } - try { - await($list, new TimeoutCancellation(1800)); - } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException $exception) { - $ctx->trigger( - Event::TimeoutCancellation, - new SystemExceptionEvent($exception), - ); - } - + $ctx->awaitFutures($list); $ctx->sendAck(); } } @@ -76,7 +64,7 @@ public function __invoke(Context $ctx, WorkflowClaim $claim, WorkflowCatch $catc /** * @return iterable */ - private function read(RedisConsume $command): iterable + private function read(Context $ctx, RedisConsume $command): iterable { /** * @psalm-var Closure(RedisConsume):iterable $fn @@ -100,9 +88,9 @@ private function read(RedisConsume $command): iterable }; /** - * @phpstan-var iterable + * @var iterable */ - return async($fn(...), $command)->await(); + return $ctx->awaitFuture(async($fn(...), $command)) ?? []; } private function autoclaim(Context $ctx, WorkflowClaim $claim, WorkflowCatch $catch, int &$lastAction): void