Updated pipeline contracts to v0.5

Author: gplanchat

composer.json

@@ -19,7 +19,7 @@
         "php": "^8.2",
         "psr/log": "^3.0",
         "php-etl/bucket": "*",
-        "php-etl/pipeline-contracts": "0.4.*",
+        "php-etl/pipeline-contracts": "0.5.*",
         "php-etl/bucket-contracts": "0.2.*"
     },
     "require-dev": {
@@ -31,7 +31,7 @@
         "rector/rector": "^0.15"
     },
     "provide": {
-        "php-etl/pipeline-implementation": "0.3.0"
+        "php-etl/pipeline-implementation": "0.5.0"
     },
     "autoload": {
         "psr-4": {
@@ -46,12 +46,13 @@
     "config": {
         "bin-dir": "bin",
         "allow-plugins": {
-            "infection/extension-installer": true
+            "infection/extension-installer": true,
+            "php-http/discovery": true
         }
     },
     "extra": {
         "branch-alias": {
-            "dev-main": "0.5.x-dev"
+            "dev-main": "0.6.x-dev"
         }
     }
 }

composer.lock

@@ -4,6 +4,7 @@
 
 namespace Kiboko\Component\Pipeline\Extractor;
 
+use Kiboko\Component\Bucket\AcceptanceResultBucket;
 use Kiboko\Contract\Pipeline\ExtractorInterface;
 
 class IteratorExtractor implements ExtractorInterface

src/Extractor/IteratorExtractor.php

@@ -14,6 +14,9 @@
 use Kiboko\Contract\Pipeline\RejectionInterface;
 use Kiboko\Contract\Pipeline\RunnableInterface;
 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\Pipeline\TransformingInterface;
 use Kiboko\Contract\Pipeline\WalkableInterface;
@@ -23,8 +26,11 @@ class Pipeline implements PipelineInterface, WalkableInterface, RunnableInterfac
     private readonly \AppendIterator $source;
     private iterable $subject;
 
-    public function __construct(private readonly PipelineRunnerInterface $runner, ?\Iterator $source = null)
-    {
+    public function __construct(
+        private readonly PipelineRunnerInterface $runner,
+        private readonly StateInterface $state,
+        ?\Iterator $source = null
+    ) {
         $this->source = new \AppendIterator();
         $this->source->append($source ?? new \EmptyIterator());
 
@@ -39,14 +45,14 @@ public function feed(...$data): void
     private function passThroughCoroutine(): \Generator
     {
         $line = yield;
-        while ($line = yield $line) {
-        }
+        while ($line = yield $line);
     }
 
     public function extract(
+        StepCodeInterface $stepCode,
         ExtractorInterface $extractor,
-        RejectionInterface $rejection,
-        StateInterface $state,
+        StepRejectionInterface $rejection,
+        StepStateInterface $state,
     ): ExtractingInterface {
         $extract = $extractor->extract();
         if (\is_array($extract)) {
@@ -84,9 +90,10 @@ public function extract(
     }
 
     public function transform(
+        StepCodeInterface $stepCode,
         TransformerInterface $transformer,
-        RejectionInterface $rejection,
-        StateInterface $state,
+        StepRejectionInterface $rejection,
+        StepStateInterface $state,
     ): TransformingInterface {
         if ($transformer instanceof FlushableInterface) {
             $iterator = new \AppendIterator();
@@ -125,9 +132,10 @@ public function transform(
     }
 
     public function load(
+        StepCodeInterface $stepCode,
         LoaderInterface $loader,
-        RejectionInterface $rejection,
-        StateInterface $state,
+        StepRejectionInterface $rejection,
+        StepStateInterface $state,
     ): LoadingInterface {
         if ($loader instanceof FlushableInterface) {
             $iterator = new \AppendIterator();
@@ -168,7 +176,11 @@ public function load(
 
     public function walk(): \Iterator
     {
+        $this->state->initialize();
+
         yield from $this->subject;
+
+        $this->state->teardown();
     }
 
     public function run(int $interval = 1000): int

src/Pipeline.php

@@ -10,6 +10,8 @@
 use Kiboko\Contract\Pipeline\PipelineRunnerInterface;
 use Kiboko\Contract\Pipeline\RejectionInterface;
 use Kiboko\Contract\Pipeline\StateInterface;
+use Kiboko\Contract\Pipeline\StepRejectionInterface;
+use Kiboko\Contract\Pipeline\StepStateInterface;
 use Psr\Log\LoggerInterface;
 use Psr\Log\LogLevel;
 use Psr\Log\NullLogger;
@@ -23,12 +25,9 @@ public function __construct(private readonly LoggerInterface $logger = new NullL
     public function run(
         \Iterator $source,
         \Generator $coroutine,
-        RejectionInterface $rejection,
-        StateInterface $state,
+        StepRejectionInterface $rejection,
+        StepStateInterface $state,
     ): \Iterator {
-        $state->initialize();
-        $rejection->initialize();
-
         $wrapper = new GeneratorWrapper();
         $wrapper->rewind($source, $coroutine);
 
@@ -69,8 +68,5 @@ public function run(
 
             $wrapper->next($source);
         }
-
-        $state->teardown();
-        $rejection->teardown();
     }
 }

src/PipelineRunner.php

@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Kiboko\Component\Pipeline;
+
+use Kiboko\Contract\Pipeline\StepCodeInterface;
+
+final class StepCode implements StepCodeInterface
+{
+    private function __construct(
+        private readonly string $reference,
+    ) {
+    }
+
+    public static function fromString(string $reference): self
+    {
+        return new self($reference);
+    }
+
+    public function __toString(): string
+    {
+        return $this->reference;
+    }
+}

src/StepCode.php

@@ -9,6 +9,8 @@
 use Kiboko\Component\Pipeline\PipelineRunner;
 use Kiboko\Contract\Pipeline\NullRejection;
 use Kiboko\Contract\Pipeline\NullState;
+use Kiboko\Contract\Pipeline\NullStepRejection;
+use Kiboko\Contract\Pipeline\NullStepState;
 use PHPUnit\Framework\TestResult;
 use Psr\Log\NullLogger;
 
@@ -115,7 +117,7 @@ public function testRun(\Iterator $source, callable $callback, array $expected):
     {
         $run = new PipelineRunner(new NullLogger());
 
-        $it = $run->run($source, $callback(), new NullRejection(), new NullState());
+        $it = $run->run($source, $callback(), new NullStepRejection(), new NullStepState());
 
         $this->assertIteration(new \ArrayIterator($expected), $it);
     }