Merge pull request #6 from php-etl/fix/pipeline-execution

Created a Proxy to fix pipeline executions in a workflow

Author: gplanchat

.github/workflows/phpstan-6.yaml

@@ -1,7 +1,7 @@
 name: PHPStan level 6
 on: push
 jobs:
-  phpstan:
+  phpstan6:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3

.github/workflows/phpstan-7.yaml

@@ -1,7 +1,7 @@
 name: PHPStan level 7
 on: push
 jobs:
-  phpstan:
+  phpstan7:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3

.github/workflows/phpstan-8.yaml

@@ -1,7 +1,7 @@
 name: PHPStan level 8
 on: push
 jobs:
-  phpstan:
+  phpstan8:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3

composer.json

@@ -44,7 +44,7 @@
     },
     "extra": {
         "branch-alias": {
-            "dev-main": "0.3.x-dev"
+            "dev-main": "0.4.x-dev"
         }
     }
 }

src/Console.php

@@ -6,8 +6,9 @@
 
 use Kiboko\Component\Action\Action;
 use Kiboko\Component\Pipeline\Pipeline;
+use Kiboko\Component\Runtime\Action\ActionRuntimeInterface;
 use Kiboko\Component\Runtime\Action\Console as ActionConsoleRuntime;
-use Kiboko\Component\Runtime\Pipeline\Console as PipelineConsoleRuntime;
+use Kiboko\Component\Runtime\Pipeline\PipelineRuntimeInterface;
 use Kiboko\Component\State;
 use Kiboko\Contract\Pipeline\PipelineRunnerInterface;
 use Kiboko\Contract\Satellite\RunnableInterface as JobRunnableInterface;
@@ -27,16 +28,16 @@ public function __construct(
         $this->state = new State\StateOutput\Workflow($output);
     }
 
-    public function loadPipeline(string $filename): PipelineConsoleRuntime
+    public function loadPipeline(string $filename): PipelineRuntimeInterface
     {
         $factory = require $filename;
 
         $pipeline = new Pipeline($this->pipelineRunner);
 
-        return $factory(new PipelineConsoleRuntime($this->output, $pipeline, $this->state->withPipeline(basename($filename))));
+        return new PipelineProxy($factory, $this->output, $pipeline, $this->state, basename($filename));
     }
 
-    public function loadAction(string $filename): ActionConsoleRuntime
+    public function loadAction(string $filename): ActionRuntimeInterface
     {
         $factory = require $filename;
 

src/PipelineProxy.php

@@ -0,0 +1,84 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Kiboko\Component\Runtime\Workflow;
+
+use Kiboko\Component\Runtime\Pipeline\Console as PipelineConsoleRuntime;
+use Kiboko\Component\Runtime\Pipeline\PipelineRuntimeInterface;
+use Kiboko\Component\State;
+use Kiboko\Contract\Pipeline\ExtractorInterface;
+use Kiboko\Contract\Pipeline\LoaderInterface;
+use Kiboko\Contract\Pipeline\PipelineInterface;
+use Kiboko\Contract\Pipeline\RejectionInterface;
+use Kiboko\Contract\Pipeline\StateInterface;
+use Kiboko\Contract\Pipeline\TransformerInterface;
+use Kiboko\Contract\Pipeline\WalkableInterface;
+use Symfony\Component\Console\Output\ConsoleOutput;
+
+class PipelineProxy implements PipelineRuntimeInterface
+{
+    /** @var list<callable> */
+    private array $queuedCalls = [];
+
+    public function __construct(
+        callable $factory,
+        private readonly ConsoleOutput $output,
+        private readonly PipelineInterface&WalkableInterface $pipeline,
+        private readonly State\StateOutput\Workflow $state,
+        private readonly string $filename,
+    ) {
+        $this->queuedCalls[] = static function (PipelineConsoleRuntime $runtime) use ($factory): void {
+            $factory($runtime);
+        };
+    }
+
+    public function extract(
+        ExtractorInterface $extractor,
+        RejectionInterface $rejection,
+        StateInterface $state,
+    ): self {
+        $this->queuedCalls[] = static function (PipelineConsoleRuntime $runtime) use ($extractor, $rejection, $state): void {
+            $runtime->extract($extractor, $rejection, $state);
+        };
+
+        return $this;
+    }
+
+    public function transform(
+        TransformerInterface $transformer,
+        RejectionInterface $rejection,
+        StateInterface $state,
+    ): self {
+        $this->queuedCalls[] = static function (PipelineConsoleRuntime $runtime) use ($transformer, $rejection, $state): void {
+            $runtime->transform($transformer, $rejection, $state);
+        };
+
+        return $this;
+    }
+
+    public function load(
+        LoaderInterface $loader,
+        RejectionInterface $rejection,
+        StateInterface $state,
+    ): self {
+        $this->queuedCalls[] = static function (PipelineConsoleRuntime $runtime) use ($loader, $rejection, $state): void {
+            $runtime->load($loader, $rejection, $state);
+        };
+
+        return $this;
+    }
+
+    public function run(int $interval = 1000): int
+    {
+        $runtime = new PipelineConsoleRuntime($this->output, $this->pipeline, $this->state->withPipeline($this->filename));
+
+        foreach ($this->queuedCalls as $queuedCall) {
+            $queuedCall($this);
+        }
+
+        $this->queuedCalls = [];
+
+        return $runtime->run($interval);
+    }
+}