✨ Add Symfony Bridge

CHANGELOG.md

@@ -12,6 +12,9 @@
 - Flow\Flow\YFlow rework
 - Add more exemples in `examples/yflow.php` to play with Y-Combinators
 - Update DX for Flow\DriverInterface : add `defer` to gain much granular control on asynchronous callbacks
+- Add Symfony Bridge
+    - Flow can now use `Flow\JobInterface` as job input
+    - new `Flow\Attribute\AsJob` attribute allows cast job on function or class and embed it's name and description
 
 ## v1.2.0
 

docs/src/content/en/docs/getting-started/job.md

@@ -0,0 +1,20 @@
+---
+title: "Job"
+description: "Job."
+lead: "Job."
+date: 2020-10-13T15:21:01+02:00
+lastmod: 2020-10-13T15:21:01+02:00
+draft: false
+images: []
+menu:
+  docs:
+    parent: "getting-started"
+weight: 10
+toc: true
+---
+
+# Job
+
+## Make your own Job
+
+You can make your custom Job by implementing `Flow\JobInterface`.

docs/src/content/en/docs/getting-started/ressources.md

@@ -93,6 +93,7 @@ Video of Y-Combinator : [https://www.youtube.com/watch?v=QSS_ZcO8Q1g](https://ww
 - Combinator : [https://github.com/loophp/combinator](https://github.com/loophp/combinator)
 - Lambda-php : [https://github.com/igorw/lambda-php](https://github.com/igorw/lambda-php)
 - Deriving the y combinator in 7 easy steps : [https://gist.github.com/igstan/388351](https://gist.github.com/igstan/388351)
+- Y combinator real life application: recursive memoization in clojure : [https://blog.klipse.tech/lambda/2016/08/10/y-combinator-app.html](https://blog.klipse.tech/lambda/2016/08/10/y-combinator-app.html)
 
 ## Messaging approach with East oriented code from [Frédéric Hardy](https://twitter.com/mageekguy)
 

examples/flow.php

@@ -17,6 +17,7 @@
 use Flow\Flow\Flow;
 use Flow\Ip;
 use Flow\IpStrategy\MaxIpStrategy;
+use Flow\Job\ClosureJob;
 
 $driver = match (random_int(1, 4)) {
     1 => new AmpDriver(),
@@ -49,7 +50,7 @@
     return new DataB($dataA->id, $d, $dataA->c);
 };
 
-$job2 = static function (DataB $dataB) use ($driver): DataC {
+$job2 = new ClosureJob(static function (DataB $dataB) use ($driver): DataC {
     printf(".* #%d - Job 2 Calculating %d * %d\n", $dataB->id, $dataB->d, $dataB->e);
 
     // simulating calculating some "heavy" operation from from 1 to 3 seconds
@@ -65,7 +66,7 @@
     printf(".* #%d - Job 2 Result for %d * %d = %d and took %.01f seconds\n", $dataB->id, $dataB->d, $dataB->e, $f, $delay);
 
     return new DataC($dataB->id, $f);
-};
+});
 
 $job3 = static function (DataC $dataC): DataD {
     printf("** #%d - Job 3 Result is %d\n", $dataC->id, $dataC->f);

src/Attribute/AsJob.php

@@ -0,0 +1,16 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Attribute;
+
+use Attribute;
+
+#[Attribute(Attribute::TARGET_CLASS | Attribute::TARGET_FUNCTION | Attribute::TARGET_METHOD)]
+class AsJob
+{
+    public function __construct(
+        public string $name = '',
+        public string $description = '',
+    ) {}
+}

src/Bridge/Symfony/DependencyInjection/Configuration.php

@@ -0,0 +1,16 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Bridge\Symfony\DependencyInjection;
+
+use Symfony\Component\Config\Definition\Builder\TreeBuilder;
+use Symfony\Component\Config\Definition\ConfigurationInterface;
+
+class Configuration implements ConfigurationInterface
+{
+    public function getConfigTreeBuilder(): TreeBuilder
+    {
+        return new TreeBuilder('flow');
+    }
+}

src/Bridge/Symfony/DependencyInjection/FlowExtension.php

@@ -0,0 +1,13 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Bridge\Symfony\DependencyInjection;
+
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\DependencyInjection\Extension\Extension;
+
+class FlowExtension extends Extension
+{
+    public function load(array $configs, ContainerBuilder $container): void {}
+}

src/Bridge/Symfony/FlowBundle.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Bridge\Symfony;
+
+use Symfony\Component\HttpKernel\Bundle\Bundle;
+
+class FlowBundle extends Bundle {}

src/Bridge/Symfony/Resources/config/services.php

@@ -0,0 +1,7 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Symfony\Component\DependencyInjection\Loader\Configurator;
+
+return static function (ContainerConfigurator $container) {};

src/Driver/AmpDriver.php

@@ -15,6 +15,7 @@
 use Flow\Event\PushEvent;
 use Flow\Exception\RuntimeException;
 use Flow\Ip;
+use Flow\JobInterface;
 use Revolt\EventLoop;
 use Revolt\EventLoop\Driver;
 use RuntimeException as NativeRuntimeException;
@@ -49,10 +50,10 @@ public function __construct(?Driver $driver = null)
     /**
      * @return Closure(TArgs): Future<TReturn>
      */
-    public function async(Closure $callback): Closure
+    public function async(Closure|JobInterface $callback): Closure
     {
         return static function (...$args) use ($callback) {
-            return async(static function (Closure $callback, array $args) {
+            return async(static function (Closure|JobInterface $callback, array $args) {
                 try {
                     return $callback(...$args, ...($args = []));
                 } catch (Throwable $exception) {
@@ -86,7 +87,7 @@ public function defer(Closure $callback): Future
 
     public function await(array &$stream): void
     {
-        $async = function (Closure $job) {
+        $async = function (Closure|JobInterface $job) {
             return function (mixed $data) use ($job) {
                 $async = $this->async($job);
 
@@ -99,7 +100,7 @@ public function await(array &$stream): void
             };
         };
 
-        $defer = function (Closure $job) {
+        $defer = function (Closure|JobInterface $job) {
             return function (Closure $map) use ($job) {
                 /** @var Closure(TReturn): mixed $map */
                 $future = $this->defer($job);

src/Driver/FiberDriver.php

@@ -16,6 +16,7 @@
 use Flow\Event\PushEvent;
 use Flow\Exception\RuntimeException;
 use Flow\Ip;
+use Flow\JobInterface;
 use Throwable;
 
 use function array_key_exists;
@@ -34,7 +35,7 @@ class FiberDriver implements DriverInterface
      */
     private array $ticks = [];
 
-    public function async(Closure $callback): Closure
+    public function async(Closure|JobInterface $callback): Closure
     {
         return static function (...$args) use ($callback) {
             return new Fiber(static function () use ($callback, $args) {
@@ -69,7 +70,7 @@ public function defer(Closure $callback): mixed
     public function await(array &$stream): void
     {
         $async = function ($isTick) use (&$fiberDatas) {
-            return function (Closure $job) use (&$fiberDatas, $isTick) {
+            return function (Closure|JobInterface $job) use (&$fiberDatas, $isTick) {
                 return function (mixed $data) use (&$fiberDatas, $isTick, $job) {
                     $async = $this->async($job);
 
@@ -97,7 +98,7 @@ public function await(array &$stream): void
         };
 
         $defer = static function ($isTick) {
-            return static function (Closure $job) use ($isTick) {
+            return static function (Closure|JobInterface $job) use ($isTick) {
                 return static function (Closure $next) use ($isTick, $job) {
                     $fiber = new Fiber(static function () use ($isTick, $job, $next) {
                         try {
@@ -136,9 +137,9 @@ public function await(array &$stream): void
                 foreach ($stream['dispatchers'] as $index => $dispatcher) {
                     $nextIp = $dispatcher->dispatch(new PullEvent(), Event::PULL)->getIp();
                     if ($nextIp !== null) {
-                        $stream['dispatchers'][$index]->dispatch(new AsyncEvent(static function (Closure $job) use ($async) {
+                        $stream['dispatchers'][$index]->dispatch(new AsyncEvent(static function (Closure|JobInterface $job) use ($async) {
                             return $async(false)($job);
-                        }, static function (Closure $job) use ($defer) {
+                        }, static function (Closure|JobInterface $job) use ($defer) {
                             return $defer(false)($job);
                         }, $stream['fnFlows'][$index]['job'], $nextIp, static function ($data) use (&$stream, $index, $nextIp) {
                             if ($data instanceof RuntimeException and array_key_exists($index, $stream['fnFlows']) and $stream['fnFlows'][$index]['errorJob'] !== null) {

src/Driver/ReactDriver.php

@@ -13,6 +13,7 @@
 use Flow\Event\PushEvent;
 use Flow\Exception\RuntimeException;
 use Flow\Ip;
+use Flow\JobInterface;
 use React\EventLoop\Loop;
 use React\EventLoop\LoopInterface;
 use React\Promise\Deferred;
@@ -46,7 +47,7 @@ public function __construct(?LoopInterface $eventLoop = null)
         $this->eventLoop = $eventLoop ?? Loop::get();
     }
 
-    public function async(Closure $callback): Closure
+    public function async(Closure|JobInterface $callback): Closure
     {
         return static function (...$args) use ($callback) {
             return async(static function () use ($callback, $args) {
@@ -81,7 +82,7 @@ public function defer(Closure $callback): Promise
 
     public function await(array &$stream): void
     {
-        $async = function (Closure $job) {
+        $async = function (Closure|JobInterface $job) {
             return function (mixed $data) use ($job) {
                 $async = $this->async($job);
 
@@ -93,7 +94,7 @@ public function await(array &$stream): void
             };
         };
 
-        $defer = function (Closure $job) {
+        $defer = function (Closure|JobInterface $job) {
             return function ($then) use ($job) {
                 $promise = $this->defer($job);
                 $promise->then($then);

src/Driver/SpatieDriver.php

@@ -15,6 +15,7 @@
 use Flow\Event\PushEvent;
 use Flow\Exception\RuntimeException;
 use Flow\Ip;
+use Flow\JobInterface;
 use RuntimeException as NativeRuntimeException;
 use Spatie\Async\Pool;
 use Throwable;
@@ -45,7 +46,7 @@ public function __construct()
         }
     }
 
-    public function async(Closure $callback): Closure
+    public function async(Closure|JobInterface $callback): Closure
     {
         return function (...$args) use ($callback) {
             return function ($onResolve) use ($callback, $args) {
@@ -67,15 +68,15 @@ public function defer(Closure $callback): mixed
 
     public function await(array &$stream): void
     {
-        $async = function (Closure $job) {
+        $async = function (Closure|JobInterface $job) {
             return function (mixed $data) use ($job) {
                 $async = $this->async($job);
 
                 return $async($data);
             };
         };
 
-        $defer = function (Closure $job) {
+        $defer = function (Closure|JobInterface $job) {
             return function (Closure $onResolve) use ($job) {
                 $this->pool->add(static function () use ($job, $onResolve) {
                     return $job($onResolve, static function ($fn, $next) {

src/Driver/SwooleDriver.php

@@ -14,6 +14,7 @@
 use Flow\Event\PushEvent;
 use Flow\Exception\RuntimeException;
 use Flow\Ip;
+use Flow\JobInterface;
 use OpenSwoole\Timer;
 use RuntimeException as NativeRuntimeException;
 use Throwable;
@@ -38,7 +39,7 @@ public function __construct()
         }
     }
 
-    public function async(Closure $callback): Closure
+    public function async(Closure|JobInterface $callback): Closure
     {
         return static function (...$args) use ($callback) {
             return static function ($onResolve) use ($callback, $args) {
@@ -61,15 +62,15 @@ public function defer(Closure $callback): mixed
 
     public function await(array &$stream): void
     {
-        $async = function (Closure $job) {
+        $async = function (Closure|JobInterface $job) {
             return function (mixed $data) use ($job) {
                 $async = $this->async($job);
 
                 return $async($data);
             };
         };
 
-        $defer = static function (Closure $job) {
+        $defer = static function (Closure|JobInterface $job) {
             return static function (Closure $onResolve) use ($job) {
                 go(static function () use ($job, $onResolve) {
                     try {

src/DriverInterface.php

@@ -13,11 +13,11 @@
 interface DriverInterface
 {
     /**
-     * #return Closure(TArgs): void when called this start async $callback.
+     * #return JobInterface<TArgs,void>|Closure(TArgs): void when called this start async $callback.
      *
-     * @param Closure(TArgs): TReturn $callback
+     * @param Closure(TArgs): TReturn|JobInterface<TArgs,TReturn> $callback
      */
-    public function async(Closure $callback): Closure;
+    public function async(Closure|JobInterface $callback): Closure;
 
     /**
      * This allow more granular control on async

src/Event/AsyncEvent.php

@@ -6,20 +6,22 @@
 
 use Closure;
 use Flow\Ip;
+use Flow\JobInterface;
 use Symfony\Contracts\EventDispatcher\Event;
 
 /**
- * @template T
+ * @template T1
  */
 final class AsyncEvent extends Event
 {
     /**
-     * @param Ip<T> $ip
+     * @param Closure|JobInterface<mixed,mixed> $job
+     * @param Ip<T1>                            $ip
      */
     public function __construct(
         private Closure $async,
         private Closure $defer,
-        private Closure $job,
+        private Closure|JobInterface $job,
         private Ip $ip,
         private Closure $callback
     ) {}
@@ -34,13 +36,16 @@ public function getDefer(): Closure
         return $this->defer;
     }
 
-    public function getJob(): Closure
+    /**
+     * @return Closure|JobInterface<mixed,mixed>
+     */
+    public function getJob(): Closure|JobInterface
     {
         return $this->job;
     }
 
     /**
-     * @return Ip<T>
+     * @return Ip<T1>
      */
     public function getIp(): Ip
     {