Merge pull request #47 from darkwood-com/1.x-dev

v1.1.5

.editorconfig

@@ -4,9 +4,12 @@ root = true # https://editorconfig.org
 end_of_line = lf
 insert_final_newline = true
 trim_trailing_whitespace = true
-indent_style = space
+indent_style = tab
 indent_size = 4
 charset = utf-8
 
 [*.md]
-max_line_length = 80
+max_line_length = 80
+
+[{*.yml,*.yaml}]
+indent_style = space

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## v1.1.5
+
+- Add Flow\FlowInterface::do notation from https://github.com/fp4php/functional
+- Update Flow\FlowInterface::fn to accept as first argument
+    - Closure : it's the job itself
+    - array : constructor arguments for Flow instanciation
+    - array (view as shape) : configuration for Flow instanciation
+    - FlowInterface : the FlowInterface instance itself
+- Update to Symfony 7.0
+
 ## v1.1.4
 
 - Add generic templating

README.md

@@ -29,11 +29,21 @@ composer require darkwood/flow
 use Flow\Flow\Flow;
 use Flow\Ip;
 
-$flow = (new Flow(fn (object $data) => $data['number'] += 1))
-    ->fn(new Flow(fn (object $data) => $data['number'] *= 2));
+class D1 {
+    public function __construct(public int $n1) {}
+}
 
-$ip = new Ip(new ArrayObject(['number' => 4]));
-$flow($ip, fn ($ip) => printf("my number %d\n", $ip->data['number'])); // display 'my number 10'
+class D2 {
+    public function __construct(public int $n2) {}
+}
+
+$flow = Flow::do(static function() {
+    yield fn (D1 $data1) => new D2($data1->n1 += 1);
+    yield fn (D2 $data2) => $data2->n2 * 2;
+});
+
+$ip = new Ip(new D1(4));
+$flow($ip, fn ($ip) => printf("my number %d\n", $ip->data->n2)); // display 'my number 10'
 ```
 
 ## Examples

composer.json

@@ -25,12 +25,12 @@
     },
     "require-dev": {
         "amphp/amp": "^3.0",
-        "openswoole/ide-helper": "^22.0",
-        "react/async": "^4.1",
-        "symfony/doctrine-messenger": "^6.3",
-        "symfony/messenger": "^6.3",
+        "openswoole/ide-helper": "^22.0.1",
+        "react/async": "^4.2",
+        "symfony/doctrine-messenger": "^7.0",
+        "symfony/messenger": "^7.0",
         "symfony/orm-pack": "^2.4",
-        "revolt/event-loop": "^1.0",
+        "revolt/event-loop": "^1.0.6",
         "spatie/async": "^1.6"
     },
     "suggest": {

docs/package.json

@@ -12,7 +12,7 @@
   "keywords": [
     "flow"
   ],
-  "version": "1.1.4",
+  "version": "1.1.5",
   "browserslist": [
     "defaults"
   ],

examples/flow.php

@@ -82,9 +82,10 @@
     $ip->data->number = null;
 };
 
-$flow = (new Flow($job1, $errorJob1, new MaxIpStrategy(2), $driver))
-    ->fn(new Flow($job2, $errorJob2, new MaxIpStrategy(2), $driver))
-;
+$flow = Flow::do(static function () use ($job1, $job2, $errorJob1, $errorJob2) {
+    yield [$job1, $errorJob1, new MaxIpStrategy(2)];
+    yield [$job2, $errorJob2, new MaxIpStrategy(2)];
+}, ['driver' => $driver]);
 
 $ipPool = new SplObjectStorage();
 

src/Exception/LogicException.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Exception;
+
+use Flow\ExceptionInterface;
+use LogicException as NativeLogicException;
+
+class LogicException extends NativeLogicException implements ExceptionInterface
+{
+}

src/Flow/Flow.php

@@ -7,13 +7,16 @@
 use Closure;
 use Flow\Driver\FiberDriver;
 use Flow\DriverInterface;
+use Flow\Exception\LogicException;
 use Flow\ExceptionInterface;
 use Flow\FlowInterface;
 use Flow\Ip;
 use Flow\IpStrategy\LinearIpStrategy;
 use Flow\IpStrategyInterface;
+use Generator;
 use SplObjectStorage;
 
+use function array_key_exists;
 use function count;
 use function is_array;
 
@@ -81,13 +84,39 @@ public function __invoke(Ip $ip, Closure $callback = null): void
         $this->nextIpJob();
     }
 
-    /**
-     * @param FlowInterface<T2> $flow
-     *
-     * @return FlowInterface<T1>
-     */
-    public function fn(FlowInterface $flow): FlowInterface
+    public static function do(callable $callable, ?array $config = null): FlowInterface
     {
+        /**
+         * @var Closure|Generator $generator
+         */
+        $generator = $callable();
+
+        if ($generator instanceof Generator) {
+            $flows = [];
+
+            while ($generator->valid()) {
+                $flow = self::flowUnwrap($generator->current(), $config);
+
+                $generator->send($flow);
+
+                $flows[] = $flow;
+            }
+
+            $return = $generator->getReturn();
+            if (!empty($return)) {
+                $flows[] = self::flowUnwrap($return, $config);
+            }
+
+            return self::flowMap($flows);
+        }
+
+        return self::flowUnwrap($generator, $config);
+    }
+
+    public function fn(array|Closure|FlowInterface $flow): FlowInterface
+    {
+        $flow = self::flowUnwrap($flow);
+
         if ($this->fnFlow) {
             $this->fnFlow->fn($flow);
         } else {
@@ -133,4 +162,59 @@ private function nextIpJob(): void
             })($ip->data);
         }
     }
+
+    /**
+     * @template TI
+     *
+     * @param array<mixed>|Closure|FlowInterface<TI> $flow
+     * @param ?array<mixed>                          $config
+     *
+     * @return FlowInterface<mixed>
+     *
+     * #param ?array{
+     *  0: Closure|array,
+     *  1?: Closure|array,
+     *  2?: IpStrategyInterface
+     *  3?: DriverInterface
+     * }|array{
+     *  "jobs"?: Closure|array,
+     *  "errorJobs"?: Closure|array,
+     *  "ipStrategy"?: IpStrategyInterface
+     *  "driver"?: DriverInterface
+     * } $config
+     */
+    private static function flowUnwrap($flow, ?array $config = null): FlowInterface
+    {
+        if ($flow instanceof Closure) {
+            return new self(...[...['jobs' => $flow], ...($config ?? [])]);
+        }
+        if (is_array($flow)) {
+            if (array_key_exists(0, $flow) || array_key_exists('jobs', $flow)) {
+                return new self(...[...$flow, ...($config ?? [])]);
+            }
+
+            return self::flowMap($flow);
+        }
+
+        return $flow;
+    }
+
+    /**
+     * @param array<FlowInterface<mixed>> $flows
+     *
+     * @return FlowInterface<mixed>
+     */
+    private static function flowMap(array $flows)
+    {
+        $flow = array_shift($flows);
+        if (null === $flow) {
+            throw new LogicException('Flow is empty');
+        }
+
+        foreach ($flows as $flowIt) {
+            $flow = $flow->fn($flowIt);
+        }
+
+        return $flow;
+    }
 }

src/Flow/FlowDecorator.php

@@ -23,21 +23,18 @@ public function __construct(private FlowInterface $flow)
     {
     }
 
-    /**
-     * @param Ip<T1> $ip
-     */
     public function __invoke(Ip $ip, Closure $callback = null): void
     {
         ($this->flow)($ip, $callback);
     }
 
-    /**
-     * @param FlowInterface<T2> $flow
-     *
-     * @return FlowInterface<T1>
-     */
-    public function fn(FlowInterface $flow): FlowInterface
+    public function fn(array|Closure|FlowInterface $flow): FlowInterface
     {
         return $this->flow->fn($flow);
     }
+
+    public static function do(callable $callable, ?array $config = null): FlowInterface
+    {
+        return Flow::do($callable, $config);
+    }
 }

src/FlowInterface.php

@@ -5,6 +5,7 @@
 namespace Flow;
 
 use Closure;
+use Generator;
 
 /**
  * @template T1
@@ -20,9 +21,59 @@ public function __invoke(Ip $ip, Closure $callback = null): void;
     /**
      * @template T2
      *
-     * @param FlowInterface<T2> $flow
+     * @param array<mixed>|Closure|FlowInterface<T2> $flow can be Closure as Job, array constructor arguments for Flow instanciation, array configuration for Flow instanciation or FlowInterface instance
+     *                                                     #param ?array{
+     *                                                     0: Closure|array,
+     *                                                     1?: Closure|array,
+     *                                                     2?: IpStrategyInterface,
+     *                                                     3?: DriverInterface
+     *                                                     }|array{
+     *                                                     "jobs"?: Closure|array,
+     *                                                     "errorJobs"?: Closure|array,
+     *                                                     "ipStrategy"?: IpStrategyInterface,
+     *                                                     "driver"?: DriverInterface
+     *                                                     }|Closure|FlowInterface<T2> $config
      *
      * @return FlowInterface<T1>
      */
-    public function fn(self $flow): self;
+    public function fn(array|Closure|self $flow): self;
+
+    /**
+     * Do-notation a.k.a. for-comprehension.
+     *
+     * Syntax sugar for sequential {@see FlowInterface::fn()} calls
+     *
+     * Syntax "$flow = yield $wrapedFlow" mean:
+     * 1) $wrapedFlow can be Closure as Job, array constructor arguments for Flow instanciation, array configuration for Flow instanciation or FlowInterface instance
+     * 2) $flow is assigned as FlowInterface instance
+     * 3) optionnaly you can return another wrapedFlow
+     *
+     * ```php
+     * $flow = Flow::do(static function() {
+     *     yield new Flow(fn($a) => $a + 1);
+     *     $flow = yield fn($b) => $b * 2;
+     *     $flow = yield $flow->fn([fn($c) => $c * 4])
+     *     return [$flow, [fn($d) => $d - 8]];
+     * });
+     * ```
+     * $config if provided will be the fallback array configuration for Flow instanciation
+     *
+     * @param callable(): Generator|Closure $callable
+     * @param ?array<mixed>                 $config
+     *
+     * #param ?array{
+     *  0: Closure|array,
+     *  1?: Closure|array,
+     *  2?: IpStrategyInterface<mixed>,
+     *  3?: DriverInterface
+     * }|array{
+     *  "jobs"?: Closure|array,
+     *  "errorJobs"?: Closure|array,
+     *  "ipStrategy"?: IpStrategyInterface<mixed>,
+     *  "driver"?: DriverInterface
+     * } $config
+     *
+     * @return FlowInterface<mixed>
+     */
+    public static function do(callable $callable, ?array $config = null): self;
 }

tests/Flow/FlowTest.php

@@ -87,6 +87,31 @@ public function testJobs(DriverInterface $driver): void
         $driver->start();
     }
 
+    /**
+     * @dataProvider provideDoCases
+     *
+     * @param DriverInterface<T1,T2>  $driver
+     * @param IpStrategyInterface<T1> $ipStrategy
+     * @param array<mixed>            $config
+     */
+    public function testDo(DriverInterface $driver, IpStrategyInterface $ipStrategy, callable $callable, ?array $config, int $resultNumber): void
+    {
+        $ip = new Ip(new ArrayObject(['number' => 0]));
+        $flow = Flow::do($callable, [
+            ...['driver' => $driver, 'ipStrategy' => $ipStrategy],
+            ...($config ?? []),
+        ]);
+
+        ($flow)($ip, static function (Ip $ip) use ($driver, $resultNumber) {
+            $driver->stop();
+
+            self::assertSame(ArrayObject::class, $ip->data::class);
+            self::assertSame($resultNumber, $ip->data['number']);
+        });
+
+        $driver->start();
+    }
+
     /**
      * @return array<array<mixed>>
      */
@@ -114,4 +139,21 @@ public static function jobProvider(): iterable
             }], 0],
         ]);
     }
+
+    /**
+     * @return array<array<mixed>>
+     */
+    public static function provideDoCases(): iterable
+    {
+        return self::matrix(static fn (DriverInterface $driver) => [
+            'simpleGenerator' => [static function () {
+                yield static function (ArrayObject $data) {
+                    $data['number'] = 5;
+                };
+                yield static function (ArrayObject $data) {
+                    $data['number'] = 10;
+                };
+            }, null, 10],
+        ]);
+    }
 }