Merge remote-tracking branch 'origin/feature/rejection-reason' into f…

…eature/upgrade-pipeline-contracts
php-etl · Nov 13, 2023 · 9839494 · 9839494
2 parents dcdcd18 + a18b984
commit 9839494
Showing 1 changed file with 29 additions and 10 deletions.
diff --git a/src/PipelineRunner.php b/src/PipelineRunner.php
@@ -4,6 +4,8 @@
 
 namespace Kiboko\Component\Pipeline;
 
+use Kiboko\Component\Bucket\RejectionResultBucket;
+use Kiboko\Component\Bucket\RejectionWithReasonResultBucket;
 use Kiboko\Contract\Bucket\AcceptanceResultBucketInterface;
 use Kiboko\Contract\Bucket\RejectionResultBucketInterface;
 use Kiboko\Contract\Bucket\ResultBucketInterface;
@@ -43,17 +45,34 @@ public function run(
             }
 
             if ($bucket instanceof RejectionResultBucketInterface) {
-                foreach ($bucket->walkRejection() as $line) {
-                    $rejection->reject($line);
-                    $state->reject();
+                if ($bucket instanceof RejectionResultBucket) {
+                    foreach ($bucket->walkRejection() as $line) {
+                        $rejection->reject($line);
+                        $state->reject();
 
-                    $this->logger->log(
-                        $this->rejectionLevel,
-                        'Some data was rejected from the pipeline',
-                        [
-                            'line' => $line,
-                        ]
-                    );
+                        $this->logger->log(
+                            $this->rejectionLevel,
+                            'Some data was rejected from the pipeline',
+                            [
+                                'line' => $line,
+                            ]
+                        );
+                    }
+                }
+
+                if ($bucket instanceof RejectionWithReasonResultBucket) {
+                    foreach ($bucket->walkRejection() as $line) {
+                        $rejection->rejectWithReason($line, $bucket->getReason());
+                        $state->reject();
+
+                        $this->logger->log(
+                            $this->rejectionLevel,
+                            'Some data was rejected from the pipeline',
+                            [
+                                'line' => $line,
+                            ]
+                        );
+                    }
                 }
             }