From 0dd4685456d895a3d68ab15f55d13dc1af67c63e Mon Sep 17 00:00:00 2001
From: Dmitriy Krivopalov <dimarik82@gmail.com>
Date: Thu, 20 Nov 2025 11:19:17 +0300
Subject: [PATCH] command async

---
 src/internal/workflow/WorkflowClaim.php | 37 ++++++++++++++-----------
 src/internal/workflow/WorkflowMain.php  | 35 +++++++++++++----------
 2 files changed, 42 insertions(+), 30 deletions(-)

diff --git a/src/internal/workflow/WorkflowClaim.php b/src/internal/workflow/WorkflowClaim.php
index 9e6e733..3a71d63 100644
--- a/src/internal/workflow/WorkflowClaim.php
+++ b/src/internal/workflow/WorkflowClaim.php
@@ -59,24 +59,29 @@ public function __invoke(Context $ctx): void
      */
     private function autoclaim(RedisStreamGroup $command): iterable
     {
-        $batch = $command->autoclaim();
-        if ($batch === null) {
-            return [];
-        }
-
-        /**
-         * @var array<array{0: non-empty-string, 1: string[]}> $src
-         */
-        $src = $batch[1] ?? [];
-        foreach ($src as [$identity, $payload]) {
-            $data = [];
-            foreach (array_chunk($payload, 2) as [$k, $v]) {
-                $data[$k] = $v;
+        $fn = static function (RedisStreamGroup $command): iterable {
+            $batch = $command->autoclaim();
+            if ($batch === null) {
+                return;
             }
 
-            yield $identity => Payload::fromPayload($data);
-        }
+            /**
+             * @var array<array{0: non-empty-string, 1: string[]}> $src
+             */
+            $src = $batch[1] ?? [];
+            foreach ($src as [$identity, $payload]) {
+                $data = [];
+                foreach (array_chunk($payload, 2) as [$k, $v]) {
+                    $data[$k] = $v;
+                }
 
-        return [];
+                yield $identity => Payload::fromPayload($data);
+            }
+        };
+
+        /**
+         * @var iterable<non-empty-string, Payload>
+         */
+        return async($fn(...), $command)->await();
     }
 }
diff --git a/src/internal/workflow/WorkflowMain.php b/src/internal/workflow/WorkflowMain.php
index 1b1812f..13adabf 100644
--- a/src/internal/workflow/WorkflowMain.php
+++ b/src/internal/workflow/WorkflowMain.php
@@ -66,22 +66,29 @@ static function () use ($workflowClaim, $ctx): void {
      */
     private function read(RedisStreamGroup $command): iterable
     {
-        $batch = $command->read();
-        if ($batch === null || $batch === []) {
-            return;
-        }
+        $fn = static function (RedisStreamGroup $command): iterable {
+            $batch = $command->read();
+            if ($batch === null || $batch === []) {
+                return;
+            }
 
-        /**
-         * @var array<array{0: non-empty-string, 1: string[]}> $src
-         */
-        $src = $batch[0][1] ?? [];
-        foreach ($src as [$identity, $payload]) {
-            $data = [];
-            foreach (array_chunk($payload, 2) as [$k, $v]) {
-                $data[$k] = $v;
+            /**
+             * @var array<array{0: non-empty-string, 1: string[]}> $src
+             */
+            $src = $batch[0][1] ?? [];
+            foreach ($src as [$identity, $payload]) {
+                $data = [];
+                foreach (array_chunk($payload, 2) as [$k, $v]) {
+                    $data[$k] = $v;
+                }
+
+                yield $identity => Payload::fromPayload($data);
             }
+        };
 
-            yield $identity => Payload::fromPayload($data);
-        }
+        /**
+         * @var iterable<non-empty-string, Payload>
+         */
+        return async($fn(...), $command)->await();
     }
 }