🔥 Cleanup Batch Async Handler

.github/workflows/ci.yaml

@@ -71,7 +71,7 @@ jobs:
         uses: shivammathur/setup-php@v2
         with:
           php-version: ${{ matrix.php }}
-          extensions: xsl, openswoole
+          extensions: xsl, openswoole-26.2.0
           tools: composer
           coverage: none # disable XDebug for tests
 
@@ -172,7 +172,7 @@ jobs:
         uses: shivammathur/setup-php@v2
         with:
           php-version: ${{ matrix.php }}
-          extensions: xsl, openswoole
+          extensions: xsl, openswoole-26.2.0
           tools: composer
           coverage: none # disable XDebug for tests
 
@@ -198,7 +198,7 @@ jobs:
         uses: shivammathur/setup-php@v2
         with:
           php-version: ${{ matrix.php }}
-          extensions: xsl, openswoole
+          extensions: xsl, openswoole-26.2.0
           tools: composer
           coverage: none # disable XDebug for tests
 

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## v1.2.6
+
+- Remove symfony messenger
+
 ## v1.2.5
 
 - Update to PHP 8.5 and Symfony 8.0

composer.json

@@ -26,13 +26,12 @@
         "symfony/dependency-injection": "^8.0",
         "symfony/event-dispatcher": "^8.0",
         "symfony/http-kernel": "^8.0",
-        "symfony/messenger": "^8.0",
         "symfony/string": "^8.0"
     },
     "require-dev": {
         "amphp/amp": "^3.1",
         "doctrine/dbal": "^4.3",
-        "openswoole/ide-helper": "^22.1",
+        "openswoole/ide-helper": "^26.2.0",
         "react/async": "^4.3",
         "spatie/async": "^1.8",
         "symfony/doctrine-messenger": "^8.0",
@@ -84,7 +83,7 @@
     },
     "config": {
         "platform": {
-            "ext-openswoole": "22.0"
+            "ext-openswoole": "26.2.0"
         }
     },
     "minimum-stability": "stable"

src/AsyncHandler/BatchAsyncHandler.php

@@ -8,19 +8,20 @@
 use Flow\Event;
 use Flow\Event\AsyncEvent;
 use Flow\Event\PoolEvent;
-use Symfony\Component\Messenger\Handler\Acknowledger;
-use Symfony\Component\Messenger\Handler\BatchHandlerInterface;
-use Symfony\Component\Messenger\Handler\BatchHandlerTrait;
-use Throwable;
+
+use function count;
 
 /**
  * @template T
  *
  * @implements AsyncHandlerInterface<T>
  */
-final class BatchAsyncHandler implements BatchHandlerInterface, AsyncHandlerInterface
+final class BatchAsyncHandler implements AsyncHandlerInterface
 {
-    use BatchHandlerTrait;
+    /**
+     * @var array<AsyncEvent<T>>
+     */
+    private array $jobs = [];
 
     /**
      * @var AsyncHandlerInterface<T>
@@ -47,43 +48,27 @@ public static function getSubscribedEvents(): array
 
     public function async(AsyncEvent $event): void
     {
-        $ack = new Acknowledger(get_debug_type($this), function (?Throwable $e = null, $event = null) {
-            $this->asyncHandler->async($event);
-        });
+        $this->jobs[] = [$event];
 
-        $this->handle($event, $ack);
+        if (!$this->shouldFlush()) {
+            return;
+        }
+
+        $jobs = $this->jobs;
+        $this->jobs = [];
+
+        foreach ($jobs as [$event]) {
+            $this->asyncHandler->async($event);
+        }
     }
 
     public function pool(PoolEvent $event): void
     {
         $this->asyncHandler->pool($event);
     }
 
-    /**
-     * PHPStan should normaly pass for method.unused
-     * https://github.com/phpstan/phpstan/issues/6039
-     * https://phpstan.org/r/8f7de023-9888-4dcb-b12c-e2fcf9547b6c.
-     *
-     * @param array{0: AsyncEvent<T>, 1: Acknowledger}[] $jobs
-     *
-     * @phpstan-ignore method.unused
-     */
-    private function process(array $jobs): void
-    {
-        foreach ($jobs as [$event, $ack]) {
-            $ack->ack($event);
-        }
-    }
-
-    /**
-     * PHPStan should normaly pass for method.unused
-     * https://github.com/phpstan/phpstan/issues/6039
-     * https://phpstan.org/r/8f7de023-9888-4dcb-b12c-e2fcf9547b6c.
-     *
-     * @phpstan-ignore method.unused
-     */
-    private function getBatchSize(): int
+    private function shouldFlush(): bool
     {
-        return $this->batchSize;
+        return $this->batchSize <= count($this->jobs);
     }
 }

src/Driver/SwooleDriver.php

@@ -119,7 +119,7 @@ public function tick($interval, Closure $callback): Closure
         $tickId = Timer::tick((int) $interval, $callback);
 
         return function () use ($tickId) {
-            Timer::clear($tickId);
+            Timer::clear($tickId); // @phpstan-ignore-line
             $this->ticks--;
         };
     }

tools/phpstan/bootstrap.php

@@ -28,8 +28,3 @@ public static function run(callable $callback) {}
         public static function sleep(int $seconds): bool {}
     }
 }
-
-// Load OpenSwoole\Timer stub file
-if (!class_exists('\OpenSwoole\Timer', false)) {
-    require_once __DIR__ . '/stubs/OpenSwoole.stub.php';
-}

tools/phpstan/composer.json

@@ -5,13 +5,12 @@
         "symfony/dependency-injection": "^8.0",
         "symfony/event-dispatcher": "^8.0",
         "symfony/http-kernel": "^8.0",
-		"symfony/messenger": "^8.0",
         "symfony/string": "^8.0"
     },
     "require-dev": {
         "amphp/amp": "^3.1",
         "doctrine/dbal": "^4.3",
-        "openswoole/ide-helper": "^22.1",
+        "openswoole/ide-helper": "^26.2.0",
         "phpstan/phpstan": "^1.12",
         "phpunit/phpunit": "^11.4",
         "react/async": "^4.3",
@@ -32,7 +31,7 @@
     },
     "config": {
         "platform": {
-            "ext-openswoole": "22.0"
+            "ext-openswoole": "26.2.0"
         }
     }
 }