Add support for amp v3

phpro · Apr 28, 2023 · d64e8fa · d64e8fa
1 parent bf8785b
commit d64e8fa
Show file tree

Hide file tree

Showing 14 changed files with 136 additions and 292 deletions.
diff --git a/.github/workflows/grumphp.yml b/.github/workflows/grumphp.yml
@@ -10,7 +10,7 @@ jobs:
         strategy:
             matrix:
                 operating-system: [ubuntu-latest, macos-latest] #windows-latest currently not working
-                php-versions: ['8.0', '8.1', '8.2']
+                php-versions: ['8.1', '8.2']
                 composer-options: ['', '--prefer-lowest']
                 composer-versions: ['composer:v2']
             fail-fast: false

diff --git a/composer.json b/composer.json
@@ -14,12 +14,11 @@
         }
     ],
     "require": {
-        "php": "^8.0",
+        "php": "^8.1",
         "ext-json": "*",
         "composer-plugin-api": "~2.0",
-        "amphp/amp": "^2.6",
-        "amphp/parallel": "^1.4",
-        "amphp/parallel-functions": "^1.1",
+        "amphp/amp": "^3.0",
+        "amphp/parallel": "^2.1",
         "doctrine/collections": "^1.6.8 || ^2.0",
         "gitonomy/gitlib": "^1.3",
         "laravel/serializable-closure": "^1.1",
@@ -39,7 +38,6 @@
         "symfony/yaml": "~5.4 || ~6.0"
     },
     "require-dev": {
-        "amphp/sync": "^v1.4",
         "brianium/paratest": "^6.4",
         "composer/composer": "^2.2.6",
         "nikic/php-parser": "~4.13",

diff --git a/src/Runner/Middleware/HandleRunnerMiddleware.php b/src/Runner/Middleware/HandleRunnerMiddleware.php
@@ -4,12 +4,10 @@
 
 namespace GrumPHP\Runner\Middleware;
 
+use Amp\Future;
 use GrumPHP\Configuration\Model\RunnerConfig;
-use function Amp\call;
-use Amp\CancelledException;
-use Amp\LazyPromise;
-use Amp\MultiReasonException;
-use function Amp\Promise\wait;
+use function Amp\async;
+use function Amp\Future\await;
 use GrumPHP\Collection\TaskResultCollection;
 use GrumPHP\Runner\Promise\MultiPromise;
 use GrumPHP\Runner\TaskHandler\TaskHandler;
@@ -37,55 +35,16 @@ public function __construct(TaskHandler $taskHandler, RunnerConfig $config)
 
     public function handle(TaskRunnerContext $context, callable $next): TaskResultCollection
     {
-        return new TaskResultCollection(
-            wait(
-                /**
-                 * @return \Generator<mixed, mixed, mixed, TaskResultInterface[]>
-                 */
-                call(function () use ($context): \Generator {
-                    /**
-                     * @var \Throwable[] $errors
-                     * @var TaskResultInterface[] $results
-                     * @psalm-suppress InvalidArrayAccess
-                     * @psalm-suppress InvalidArrayOffset
-                     */
-                    [$errors, $results] = yield MultiPromise::cancelable(
-                        $this->handleTasks($context),
-                        function (TaskResultInterface $result) {
-                            return $this->config->stopOnFailure() && $result->isBlocking();
-                        }
-                    );
-
-                    // Filter out canceled items:
-                    $errors = array_filter($errors, function (\Throwable $error): bool {
-                        return !$error instanceof CancelledException;
-                    });
-
-                    if ($errors) {
-                        throw new MultiReasonException($errors);
-                    }
+        // TODO : CANCELLATION based on return $this->config->stopOnFailure() && $result->isBlocking();
 
-                    return $results;
-                })
+        return new TaskResultCollection(
+            await(
+                array_map(
+                    /** @return Future<TaskResultInterface> */
+                    fn (TaskInterface $task): Future => $this->taskHandler->handle($task, $context),
+                    $context->getTasks()->toArray()
+                )
             )
         );
     }
-
-    /**
-     * @return array<int, LazyPromise<TaskResultInterface>>
-     */
-    private function handleTasks(TaskRunnerContext $context): array
-    {
-        return array_map(
-            /**
-             * @return LazyPromise<TaskResultInterface>
-             */
-            function (TaskInterface $task) use ($context) : LazyPromise {
-                return new LazyPromise(function () use ($task, $context) {
-                    return $this->taskHandler->handle($task, $context);
-                });
-            },
-            $context->getTasks()->toArray()
-        );
-    }
 }
diff --git a/src/Runner/Parallel/PoolFactory.php b/src/Runner/Parallel/PoolFactory.php
@@ -4,26 +4,28 @@
 
 namespace GrumPHP\Runner\Parallel;
 
-use Amp\Parallel\Worker\DefaultPool;
-use Amp\Parallel\Worker\Pool;
+use Amp\Parallel\Worker\ContextWorkerPool;
+use Amp\Parallel\Worker\WorkerPool;
 use GrumPHP\Configuration\Model\ParallelConfig;
 
 class PoolFactory
 {
-    /**
-     * @var ParallelConfig
-     */
-    private $config;
+    private ParallelConfig $config;
+    private ?WorkerPool $pool = null;
 
     public function __construct(ParallelConfig $config)
     {
         $this->config = $config;
     }
 
-    public function create(): Pool
+    public function createShared(): WorkerPool
     {
-        return new DefaultPool(
-            $this->config->getMaxWorkers()
-        );
+        if (!$this->pool) {
+            $this->pool = new ContextWorkerPool(
+                $this->config->getMaxWorkers()
+            );
+        }
+
+        return $this->pool;
     }
 }
diff --git a/src/Runner/Parallel/SerializedClosureTask.php b/src/Runner/Parallel/SerializedClosureTask.php
@@ -0,0 +1,51 @@
+<?php
+declare(strict_types=1);
+
+namespace GrumPHP\Runner\Parallel;
+
+use Amp\Cancellation;
+use Amp\Parallel\Worker\Task;
+use Amp\Sync\Channel;
+use Laravel\SerializableClosure\SerializableClosure;
+
+/**
+ * @template T
+ */
+class SerializedClosureTask implements Task
+{
+    /**
+     * @param (\Closure(): T) $closure
+     */
+    public function __construct(
+        private string $serializedClosure
+    ) {
+    }
+
+    /**
+     * @template O
+     * @param \Closure(): O $closure
+     * @return self<O>
+     */
+    public static function fromClosure(\Closure $closure): self
+    {
+        return new self(serialize(SerializableClosure::unsigned($closure)));
+    }
+
+    /**
+     * @return T
+     */
+    public function run(Channel $channel, Cancellation $cancellation): mixed
+    {
+        $callable = \unserialize($this->serializedClosure, ['allowed_classes' => true]);
+
+        if ($callable instanceof \__PHP_Incomplete_Class) {
+            throw new \Error('When using a class instance as a callable, the class must be autoloadable');
+        }
+
+        if (!$callable instanceof \Closure) {
+            throw new \Error('This task can only deal with serialized Closures. You passed '.get_debug_type($callable));
+        }
+
+        return $callable();
+    }
+}
diff --git a/src/Runner/TaskHandler/Middleware/ErrorHandlingTaskHandlerMiddleware.php b/src/Runner/TaskHandler/Middleware/ErrorHandlingTaskHandlerMiddleware.php
@@ -4,9 +4,8 @@
 
 namespace GrumPHP\Runner\TaskHandler\Middleware;
 
-use Amp\Parallel\Worker\TaskFailureException;
-use function Amp\call;
-use Amp\Promise;
+use function Amp\async;
+use Amp\Future;
 use GrumPHP\Exception\PlatformException;
 use GrumPHP\Runner\TaskResult;
 use GrumPHP\Runner\TaskResultInterface;
@@ -19,8 +18,8 @@ public function handle(
         TaskInterface $task,
         TaskRunnerContext $runnerContext,
         callable $next
-    ): Promise {
-        return call(
+    ): Future {
+        return async(
             static function () use ($task, $runnerContext): TaskResultInterface {
                 $taskContext = $runnerContext->getTaskContext();
                 try {

diff --git a/src/Runner/TaskHandler/Middleware/EventDispatchingTaskHandlerMiddleware.php b/src/Runner/TaskHandler/Middleware/EventDispatchingTaskHandlerMiddleware.php
@@ -4,8 +4,8 @@
 
 namespace GrumPHP\Runner\TaskHandler\Middleware;
 
-use function Amp\call;
-use Amp\Promise;
+use function Amp\async;
+use Amp\Future;
 use GrumPHP\Event\Dispatcher\EventDispatcherInterface;
 use GrumPHP\Event\TaskEvent;
 use GrumPHP\Event\TaskEvents;
@@ -27,18 +27,14 @@ public function __construct(EventDispatcherInterface $eventDispatcher)
         $this->eventDispatcher = $eventDispatcher;
     }
 
-    public function handle(TaskInterface $task, TaskRunnerContext $runnerContext, callable $next): Promise
+    public function handle(TaskInterface $task, TaskRunnerContext $runnerContext, callable $next): Future
     {
-        return call(
-            /**
-             * @return \Generator<mixed, Promise<TaskResultInterface>, mixed, TaskResultInterface>
-             */
-            function () use ($task, $runnerContext, $next): \Generator {
+        return async(
+            function () use ($task, $runnerContext, $next): TaskResultInterface {
                 $taskContext = $runnerContext->getTaskContext();
                 $this->eventDispatcher->dispatch(new TaskEvent($task, $taskContext), TaskEvents::TASK_RUN);
 
-                /** @var TaskResultInterface $result */
-                $result = yield $next($task, $runnerContext);
+                $result = $next($task, $runnerContext)->await();
 
                 if ($result->isSkipped()) {
                     $this->eventDispatcher->dispatch(new TaskEvent($task, $taskContext), TaskEvents::TASK_SKIPPED);

diff --git a/src/Runner/TaskHandler/Middleware/MemoizedResultsTaskHandlerMiddleware.php b/src/Runner/TaskHandler/Middleware/MemoizedResultsTaskHandlerMiddleware.php
@@ -4,8 +4,8 @@
 
 namespace GrumPHP\Runner\TaskHandler\Middleware;
 
-use function Amp\call;
-use Amp\Promise;
+use function Amp\async;
+use Amp\Future;
 use GrumPHP\Runner\MemoizedTaskResultMap;
 use GrumPHP\Runner\TaskResult;
 use GrumPHP\Runner\TaskResultInterface;
@@ -24,16 +24,12 @@ public function __construct(MemoizedTaskResultMap $resultMap)
         $this->resultMap = $resultMap;
     }
 
-    public function handle(TaskInterface $task, TaskRunnerContext $runnerContext, callable $next): Promise
+    public function handle(TaskInterface $task, TaskRunnerContext $runnerContext, callable $next): Future
     {
-        return call(
-            /**
-             * @return \Generator<mixed, Promise<TaskResultInterface>, mixed, TaskResultInterface>
-             */
-            function () use ($task, $runnerContext, $next) : \Generator {
+        return async(
+            function () use ($task, $runnerContext, $next) : TaskResultInterface {
                 try {
-                    /** @var TaskResultInterface $result */
-                    $result = yield $next($task, $runnerContext);
+                    $result = $next($task, $runnerContext)->await();
                 } catch (\Throwable $error) {
                     $result = TaskResult::createFailed($task, $runnerContext->getTaskContext(), $error->getMessage());
                 }

diff --git a/src/Runner/TaskHandler/Middleware/NonBlockingTaskHandlerMiddleware.php b/src/Runner/TaskHandler/Middleware/NonBlockingTaskHandlerMiddleware.php
@@ -4,8 +4,8 @@
 
 namespace GrumPHP\Runner\TaskHandler\Middleware;
 
-use function Amp\call;
-use Amp\Promise;
+use function Amp\async;
+use Amp\Future;
 use GrumPHP\Runner\TaskResult;
 use GrumPHP\Runner\TaskResultInterface;
 use GrumPHP\Runner\TaskRunnerContext;
@@ -17,14 +17,10 @@ public function handle(
         TaskInterface $task,
         TaskRunnerContext $runnerContext,
         callable $next
-    ): Promise {
-        return call(
-            /**
-             * @return \Generator<mixed, Promise<TaskResultInterface>, mixed, TaskResultInterface>
-             */
-            static function () use ($task, $runnerContext, $next): \Generator {
-                /** @var TaskResultInterface $result */
-                $result = yield $next($task, $runnerContext);
+    ): Future {
+        return async(
+            static function () use ($task, $runnerContext, $next): TaskResultInterface {
+                $result = $next($task, $runnerContext)->await();
 
                 if ($result->isPassed() || $result->isSkipped() || $task->getConfig()->getMetadata()->isBlocking()) {
                     return $result;