Updated code to be compatible with php-etl/pipeline-contracts:0.5 and php-etl/satellite-contracts:0.1

Author: gplanchat

composer.json

@@ -21,7 +21,6 @@
         "php-etl/pipeline-contracts": "0.5.*",
         "php-etl/satellite-contracts": "0.1.*",
         "php-etl/pipeline": "*",
-        "php-etl/console-state": "*",
         "php-etl/pipeline-console-runtime": "*",
         "php-etl/action-console-runtime": "*",
         "php-etl/action": "*",

composer.lock

@@ -6,7 +6,6 @@
 
 use Kiboko\Component\Action\Action;
 use Kiboko\Component\Runtime\Action\Console as ActionConsoleRuntime;
-use Kiboko\Component\State;
 use Kiboko\Contract\Satellite\CodeInterface;
 use Kiboko\Contract\Satellite\RunnableInterface;
 use Symfony\Component\Console\Output\ConsoleOutput;
@@ -19,7 +18,7 @@ class ActionProxy implements RunnableInterface
     public function __construct(
         callable $factory,
         private readonly ConsoleOutput $output,
-        private readonly State\StateOutput\Workflow $state,
+        private readonly Workflow $state,
         private readonly CodeInterface $code,
     ) {
         $this->queuedCalls[] = static function (ActionConsoleRuntime $runtime) use ($factory): void {

src/ActionProxy.php

@@ -4,28 +4,24 @@
 
 namespace Kiboko\Component\Runtime\Workflow;
 
-use Kiboko\Component\Action\Action;
-use Kiboko\Component\Pipeline\Pipeline;
 use Kiboko\Component\Runtime\Pipeline\PipelineRuntimeInterface;
-use Kiboko\Component\State;
 use Kiboko\Contract\Pipeline\PipelineRunnerInterface;
 use Kiboko\Contract\Satellite\CodeInterface;
 use Kiboko\Contract\Satellite\RunnableInterface;
-use Kiboko\Contract\Satellite\RunnableInterface as JobRunnableInterface;
 use Symfony\Component\Console\Output\ConsoleOutput;
 
 final class Console implements WorkflowRuntimeInterface
 {
-    private readonly State\StateOutput\Workflow $state;
+    private readonly Workflow $state;
 
-    /** @var list<JobRunnableInterface> */
+    /** @var list<RunnableInterface> */
     private array $jobs = [];
 
     public function __construct(
         private readonly ConsoleOutput $output,
         private readonly PipelineRunnerInterface $pipelineRunner,
     ) {
-        $this->state = new State\StateOutput\Workflow($output);
+        $this->state = new Workflow($output);
     }
 
     public function loadPipeline(CodeInterface $job, string $filename): PipelineRuntimeInterface
@@ -42,18 +38,18 @@ public function loadAction(CodeInterface $job, string $filename): RunnableInterf
         return new ActionProxy($factory, $this->output, $this->state, $job);
     }
 
-    public function job(JobRunnableInterface $job): self
+    public function job(CodeInterface $job, RunnableInterface $runnable): self
     {
-        $this->jobs[] = $job;
+        $this->jobs[(string) $job] = [$job, $runnable];
 
         return $this;
     }
 
     public function run(int $interval = 1000): int
     {
         $count = 0;
-        foreach ($this->jobs as $job) {
-            $count = $job->run($interval);
+        foreach ($this->jobs as [$job, $runnable]) {
+            $count = $runnable->run($interval);
         }
 
         return $count;

src/Console.php

@@ -6,13 +6,16 @@
 
 use Kiboko\Component\Pipeline\Pipeline;
 use Kiboko\Component\Runtime\Pipeline\Console as PipelineConsoleRuntime;
+use Kiboko\Component\Runtime\Pipeline\MemoryState;
 use Kiboko\Component\Runtime\Pipeline\PipelineRuntimeInterface;
-use Kiboko\Component\State;
 use Kiboko\Contract\Pipeline\ExtractorInterface;
 use Kiboko\Contract\Pipeline\LoaderInterface;
 use Kiboko\Contract\Pipeline\PipelineRunnerInterface;
 use Kiboko\Contract\Pipeline\RejectionInterface;
 use Kiboko\Contract\Pipeline\StateInterface;
+use Kiboko\Contract\Pipeline\StepCodeInterface;
+use Kiboko\Contract\Pipeline\StepRejectionInterface;
+use Kiboko\Contract\Pipeline\StepStateInterface;
 use Kiboko\Contract\Pipeline\TransformerInterface;
 use Kiboko\Contract\Satellite\CodeInterface;
 use Symfony\Component\Console\Output\ConsoleOutput;
@@ -26,7 +29,7 @@ public function __construct(
         callable $factory,
         private readonly ConsoleOutput $output,
         private readonly PipelineRunnerInterface $pipelineRunner,
-        private readonly State\StateOutput\Workflow $state,
+        private readonly Workflow $state,
         private readonly CodeInterface $code,
     ) {
         $this->queuedCalls[] = static function (PipelineConsoleRuntime $runtime) use ($factory): void {
@@ -35,36 +38,39 @@ public function __construct(
     }
 
     public function extract(
+        StepCodeInterface $step,
         ExtractorInterface $extractor,
-        RejectionInterface $rejection,
-        StateInterface $state,
+        StepRejectionInterface $rejection,
+        StepStateInterface $state,
     ): self {
-        $this->queuedCalls[] = static function (PipelineConsoleRuntime $runtime) use ($extractor, $rejection, $state): void {
-            $runtime->extract($extractor, $rejection, $state);
+        $this->queuedCalls[] = static function (PipelineConsoleRuntime $runtime) use ($step, $extractor, $rejection, $state): void {
+            $runtime->extract($step, $extractor, $rejection, $state);
         };
 
         return $this;
     }
 
     public function transform(
+        StepCodeInterface $step,
         TransformerInterface $transformer,
-        RejectionInterface $rejection,
-        StateInterface $state,
+        StepRejectionInterface $rejection,
+        StepStateInterface $state,
     ): self {
-        $this->queuedCalls[] = static function (PipelineConsoleRuntime $runtime) use ($transformer, $rejection, $state): void {
-            $runtime->transform($transformer, $rejection, $state);
+        $this->queuedCalls[] = static function (PipelineConsoleRuntime $runtime) use ($step, $transformer, $rejection, $state): void {
+            $runtime->transform($step, $transformer, $rejection, $state);
         };
 
         return $this;
     }
 
     public function load(
+        StepCodeInterface $step,
         LoaderInterface $loader,
-        RejectionInterface $rejection,
-        StateInterface $state,
+        StepRejectionInterface $rejection,
+        StepStateInterface $state,
     ): self {
-        $this->queuedCalls[] = static function (PipelineConsoleRuntime $runtime) use ($loader, $rejection, $state): void {
-            $runtime->load($loader, $rejection, $state);
+        $this->queuedCalls[] = static function (PipelineConsoleRuntime $runtime) use ($step, $loader, $rejection, $state): void {
+            $runtime->load($step, $loader, $rejection, $state);
         };
 
         return $this;
@@ -73,7 +79,7 @@ public function load(
     public function run(int $interval = 1000): int
     {
         $state = $this->state->withPipeline((string) $this->code);
-        $pipeline = new Pipeline($this->pipelineRunner, new State\MemoryState());
+        $pipeline = new Pipeline($this->pipelineRunner, new MemoryState());
 
         $runtime = new PipelineConsoleRuntime($this->output, $pipeline, $state);
 

src/PipelineProxy.php

@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Kiboko\Component\Runtime\Workflow;
+
+use Kiboko\Component\Runtime\Action\Action;
+use Kiboko\Component\Runtime\Pipeline\Pipeline;
+use Symfony\Component\Console\Output\ConsoleOutput;
+
+final class Workflow
+{
+    /** @var list<Action|Pipeline> */
+    private array $jobs = [];
+    private string $index = 'A';
+
+    public function __construct(
+        private readonly ConsoleOutput $output,
+    ) {
+    }
+
+    public function withPipeline(string $label): Pipeline
+    {
+        return $this->jobs[] = new Pipeline($this->output, $this->index++, $label);
+    }
+
+    public function withAction(string $label): Action
+    {
+        return $this->jobs[] = new Action($this->output, $this->index++, $label);
+    }
+
+    public function update(): void
+    {
+        foreach ($this->jobs as $job) {
+            $job->update();
+        }
+    }
+}

src/Workflow.php

@@ -4,7 +4,7 @@
 
 namespace Kiboko\Component\Runtime\Workflow;
 
-use Kiboko\Contract\Pipeline\SchedulingInterface;
 use Kiboko\Contract\Satellite\RunnableInterface;
+use Kiboko\Contract\Satellite\SchedulingInterface;
 
 interface WorkflowRuntimeInterface extends SchedulingInterface, RunnableInterface {}