Merge branch '0.9' of github.com:chevere/workflow into 0.9

README.md

@@ -467,13 +467,6 @@ $run = run(
     image: '/path/to/image-to-upload.png',
     savePath: '/path/to/storage/'
 );
-
-// Alternative syntax
-$variables = [
-    'image' => '/path/to/image-to-upload.png',
-    'savePath' => '/path/to/storage/'
-];
-$run = run($workflow, ...$variables);
 ```
 
 Use `getReturn` to retrieve a job response as a `CastArgument` object which can be used to get a typed response.
@@ -482,6 +475,70 @@ Use `getReturn` to retrieve a job response as a `CastArgument` object which can
 $thumbFile = $run->getReturn('thumb')->string();
 ```
 
+### Sync vs Async
+
+Run live example: `php demo/sync-vs-async.php` - [view source](./demo/sync-vs-async.php)
+
+For this example you can compare the execution time between synchronous and asynchronous jobs. The example fetches the content of three URLs using `FetchUrl` action.
+
+```php
+use Chevere\Demo\Actions\FetchUrl;
+use function Chevere\Workflow\async;
+use function Chevere\Workflow\run;
+use function Chevere\Workflow\sync;
+use function Chevere\Workflow\variable;
+use function Chevere\Workflow\workflow;
+
+$sync = workflow(
+    php: sync(
+        new FetchUrl(),
+        url: variable('php'),
+    ),
+    github: sync(
+        new FetchUrl(),
+        url: variable('github'),
+    ),
+    chevere: sync(
+        new FetchUrl(),
+        url: variable('chevere'),
+    ),
+);
+$async = workflow(
+    php: async(
+        new FetchUrl(),
+        url: variable('php'),
+    ),
+    github: async(
+        new FetchUrl(),
+        url: variable('github'),
+    ),
+    chevere: async(
+        new FetchUrl(),
+        url: variable('chevere'),
+    ),
+);
+$variables = [
+    'php' => 'https://www.php.net',
+    'github' => 'https://github.com/chevere/workflow',
+    'chevere' => 'https://chevere.org',
+];
+$time = microtime(true);
+$run = run($sync, ...$variables);
+$time = microtime(true) - $time;
+echo "Time sync: {$time}\n";
+$time = microtime(true);
+$run = run($async, ...$variables);
+$time = microtime(true) - $time;
+echo "Time async: {$time}\n";
+```
+
+When running sync (blocking) jobs the execution time is higher than async (non-blocking) jobs. This is because async jobs run in parallel.
+
+```plain
+Time sync: 2.5507028102875
+Time async: 1.5810508728027
+```
+
 ### Conditional jobs
 
 Run live example: `php demo/run-if.php` - [view source](./demo/run-if.php)

composer.json

@@ -13,7 +13,7 @@
     ],
     "require": {
         "php": "^8.1",
-        "amphp/parallel": "^1.4",
+        "amphp/parallel": "^2.2",
         "chevere/action": "^1.0.0",
         "chevere/data-structure": "^1.0.1",
         "chevere/parameter": "^1.0.x-dev",

demo/Actions/FetchUrl.php

@@ -0,0 +1,30 @@
+<?php
+
+/*
+ * This file is part of Chevere.
+ *
+ * (c) Rodolfo Berrios <rodolfo@chevere.org>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace Chevere\Demo\Actions;
+
+use Chevere\Action\Action;
+use RuntimeException;
+
+class FetchUrl extends Action
+{
+    protected function main(string $url): string
+    {
+        $content = file_get_contents($url);
+        if ($content === false) {
+            throw new RuntimeException('Error fetching URL');
+        }
+
+        return $content;
+    }
+}

demo/image-resize.php

@@ -55,7 +55,7 @@
 $graph = $run->workflow()->jobs()->graph()->toArray();
 echo "Workflow graph:\n";
 foreach ($graph as $level => $jobs) {
-    echo " {$level}: [" . implode('|', $jobs) . "]\n";
+    echo " {$level}: " . implode('|', $jobs) . "\n";
 }
 echo <<<PLAIN
 thumbFile: {$run->getReturn('thumb')->string()}

demo/sync-vs-async.php

@@ -0,0 +1,66 @@
+<?php
+
+/*
+ * This file is part of Chevere.
+ *
+ * (c) Rodolfo Berrios <rodolfo@chevere.org>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+use Chevere\Demo\Actions\FetchUrl;
+use function Chevere\Workflow\async;
+use function Chevere\Workflow\run;
+use function Chevere\Workflow\sync;
+use function Chevere\Workflow\variable;
+use function Chevere\Workflow\workflow;
+
+require 'loader.php';
+
+/*
+php demo/sync-vs-async.php
+*/
+$sync = workflow(
+    php: sync(
+        new FetchUrl(),
+        url: variable('php'),
+    ),
+    github: sync(
+        new FetchUrl(),
+        url: variable('github'),
+    ),
+    chevere: sync(
+        new FetchUrl(),
+        url: variable('chevere'),
+    ),
+);
+$async = workflow(
+    php: async(
+        new FetchUrl(),
+        url: variable('php'),
+    ),
+    github: async(
+        new FetchUrl(),
+        url: variable('github'),
+    ),
+    chevere: async(
+        new FetchUrl(),
+        url: variable('chevere'),
+    ),
+);
+$variables = [
+    'php' => 'https://www.php.net',
+    'github' => 'https://github.com/chevere/workflow',
+    'chevere' => 'https://chevere.org',
+];
+$time = microtime(true);
+$run = run($sync, ...$variables);
+$time = microtime(true) - $time;
+echo "Time sync: {$time}\n";
+$time = microtime(true);
+$run = run($async, ...$variables);
+$time = microtime(true) - $time;
+echo "Time async: {$time}\n";

src/CallableTask.php

@@ -0,0 +1,45 @@
+<?php
+
+/*
+ * This file is part of Chevere.
+ *
+ * (c) Rodolfo Berrios <rodolfo@chevere.org>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace Chevere\Workflow;
+
+use Amp\Cancellation;
+use Amp\Parallel\Worker\Task;
+use Amp\Sync\Channel;
+
+/**
+ * @template-implements Task<mixed, never, never>
+ */
+final class CallableTask implements Task
+{
+    private string $callable;
+
+    /**
+     * @var array<mixed>
+     */
+    private array $arguments;
+
+    public function __construct(
+        string $callable,
+        mixed ...$arguments
+    ) {
+        $this->callable = $callable;
+        $this->arguments = $arguments;
+    }
+
+    public function run(Channel $channel, Cancellation $cancellation): mixed
+    {
+        // @phpstan-ignore-next-line
+        return ($this->callable)(...$this->arguments);
+    }
+}

src/Runner.php

@@ -13,7 +13,7 @@
 
 namespace Chevere\Workflow;
 
-use Amp\Promise;
+use Amp\Parallel\Worker\Execution;
 use Chevere\Action\Interfaces\ActionInterface;
 use Chevere\Parameter\Interfaces\CastInterface;
 use Chevere\Workflow\Interfaces\JobInterface;
@@ -24,9 +24,8 @@
 use InvalidArgumentException;
 use OutOfBoundsException;
 use Throwable;
-use function Amp\Parallel\Worker\enqueueCallable;
-use function Amp\Promise\all;
-use function Amp\Promise\wait;
+use function Amp\Future\await;
+use function Amp\Parallel\Worker\submit;
 use function Chevere\Message\message;
 use function Chevere\Parameter\cast;
 
@@ -54,9 +53,12 @@ public function withRun(): RunnerInterface
 
                 continue;
             }
-            $promises = $new->getPromises($node);
+            $executions = $new->getExecutions($node);
             /** @var RunnerInterface[] $responses */
-            $responses = wait(all($promises));
+            $responses = await(array_map(
+                fn (Execution $e) => $e->getFuture(),
+                $executions,
+            ));
             foreach ($responses as $runner) {
                 $new->merge($new, $runner);
             }
@@ -178,20 +180,22 @@ private function addJobSkip(string $name): void
 
     /**
      * @param array<string> $queue
-     * @return array<Promise<mixed>>
+     * @return array<Execution<mixed, never, never>>
      */
-    private function getPromises(array $queue): array
+    private function getExecutions(array $queue): array
     {
-        $promises = [];
+        $return = [];
         foreach ($queue as $job) {
-            $promises[] = enqueueCallable(
-                'Chevere\\Workflow\\runnerForJob',
-                $this,
-                $job,
+            $return[] = submit(
+                new CallableTask(
+                    'Chevere\\Workflow\\runnerForJob',
+                    $this,
+                    $job,
+                )
             );
         }
 
-        return $promises;
+        return $return;
     }
 
     private function merge(self $self, RunnerInterface $runner): void