Removed Pipeline::setSource

Author: norberttech

src/core/etl/src/Flow/ETL/DataFrame.php

@@ -39,16 +39,7 @@ final class DataFrame
 
     public function __construct(private Pipeline $pipeline, Config|FlowContext $context)
     {
-        $this->context = $context instanceof  FlowContext ? $context : new FlowContext($context);
-    }
-
-    /**
-     * @lazy
-     * @param callable(Pipeline $pipeline, FlowContext $context) : DataFrame $callback
-     */
-    public function rebuild(callable $callback) : self
-    {
-        return $callback($this->pipeline, $this->context);
+        $this->context = $context instanceof FlowContext ? $context : new FlowContext($context);
     }
 
     /**
@@ -614,6 +605,16 @@ public function printSchema(?int $limit = 20, Schema\SchemaFormatter $formatter
         $clone->run();
     }
 
+    /**
+     * @lazy
+     *
+     * @param callable(Pipeline $pipeline, FlowContext $context) : DataFrame $callback
+     */
+    public function rebuild(callable $callback) : self
+    {
+        return $callback($this->pipeline, $this->context);
+    }
+
     /**
      * @lazy
      */
@@ -772,8 +773,7 @@ public function sortBy(Reference ...$entries) : self
             ->cache($this->context->config->id())
             ->run();
 
-        $this->pipeline = new SynchronousPipeline();
-        $this->pipeline->setSource($this->context->config->externalSort()->sortBy(...$entries));
+        $this->pipeline = new SynchronousPipeline($this->context->config->externalSort()->sortBy(...$entries));
 
         return $this;
     }

src/core/etl/src/Flow/ETL/DataFrame/GroupedDataFrame.php

@@ -28,17 +28,17 @@ public function pivot(Reference $ref) : self
         return $this;
     }
 
-    public function toDataFrame(): DataFrame
+    public function toDataFrame() : DataFrame
     {
-        return $this->df->rebuild(function (Pipeline $pipeline, FlowContext $context): DataFrame {
+        return $this->df->rebuild(function (Pipeline $pipeline, FlowContext $context) : DataFrame {
             return new DataFrame(
                 new Pipeline\LinkedPipeline(new Pipeline\GroupByPipeline($this->groupBy, $pipeline), new Pipeline\SynchronousPipeline()),
                 $context
             );
         });
     }
 
-    public function toDF(): DataFrame
+    public function toDF() : DataFrame
     {
         return $this->toDataFrame();
     }

src/core/etl/src/Flow/ETL/Flow.php

@@ -28,7 +28,7 @@ public static function setUp(ConfigBuilder|Config $config) : self
     public function extract(Extractor $extractor) : DataFrame
     {
         return new DataFrame(
-            (new SynchronousPipeline())->setSource($extractor),
+            (new SynchronousPipeline($extractor)),
             $this->config
         );
     }
@@ -41,7 +41,7 @@ public function from(Extractor $extractor) : DataFrame
     public function process(Rows ...$rows) : DataFrame
     {
         return new DataFrame(
-            (new SynchronousPipeline())->setSource(new ProcessExtractor(...$rows)),
+            (new SynchronousPipeline(new ProcessExtractor(...$rows))),
             $this->config
         );
     }

src/core/etl/src/Flow/ETL/Pipeline.php

@@ -24,7 +24,5 @@ public function pipes() : Pipes;
      */
     public function process(FlowContext $context) : \Generator;
 
-    public function setSource(Extractor $extractor) : self;
-
     public function source() : Extractor;
 }

src/core/etl/src/Flow/ETL/Pipeline/BatchingPipeline.php

@@ -10,7 +10,6 @@
 
 final class BatchingPipeline implements Pipeline
 {
-
     /**
      * @param Pipeline $pipeline
      * @param int<1, max> $size
@@ -51,13 +50,6 @@ public function process(FlowContext $context) : \Generator
         return chunks_from(from_pipeline($this->pipeline), $this->size)->extract($context);
     }
 
-    public function setSource(Extractor $extractor) : self
-    {
-        $this->pipeline->setSource($extractor);
-
-        return $this;
-    }
-
     public function source() : Extractor
     {
         return $this->pipeline->source();

src/core/etl/src/Flow/ETL/Pipeline/CachingPipeline.php

@@ -4,7 +4,6 @@
 
 namespace Flow\ETL\Pipeline;
 
-use function Flow\ETL\DSL\from_rows;
 use Flow\ETL\{Extractor, FlowContext, Loader, Pipeline, Transformer};
 
 final class CachingPipeline implements Pipeline
@@ -49,13 +48,6 @@ public function process(FlowContext $context) : \Generator
         }
     }
 
-    public function setSource(Extractor $extractor) : Pipeline
-    {
-        $this->pipeline->setSource($extractor);
-
-        return $this;
-    }
-
     public function source() : Extractor
     {
         return $this->pipeline->source();

src/core/etl/src/Flow/ETL/Pipeline/CollectingPipeline.php

@@ -4,7 +4,6 @@
 
 namespace Flow\ETL\Pipeline;
 
-use function Flow\ETL\DSL\from_rows;
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\{Extractor, FlowContext, Loader, Pipeline, Rows, Transformer};
 
@@ -55,13 +54,6 @@ public function process(FlowContext $context) : \Generator
         yield $rows;
     }
 
-    public function setSource(Extractor $extractor) : self
-    {
-        $this->pipeline->setSource($extractor);
-
-        return $this;
-    }
-
     public function source() : Extractor
     {
         return $this->pipeline->source();

src/core/etl/src/Flow/ETL/Pipeline/GroupByPipeline.php

@@ -46,13 +46,6 @@ public function process(FlowContext $context) : \Generator
         yield $this->groupBy->result($context);
     }
 
-    public function setSource(Extractor $extractor) : self
-    {
-        $this->pipeline->setSource($extractor);
-
-        return $this;
-    }
-
     public function source() : Extractor
     {
         return $this->pipeline->source();

src/core/etl/src/Flow/ETL/Pipeline/LinkedPipeline.php

@@ -8,10 +8,12 @@
 
 final class LinkedPipeline implements OverridingPipeline, Pipeline
 {
+    private readonly Pipeline $nextPipeline;
+
     public function __construct(
         private readonly Pipeline $pipeline,
-        private readonly Pipeline $nextPipeline
     ) {
+        $this->nextPipeline = new SynchronousPipeline(new Extractor\PipelineExtractor($this->pipeline));
     }
 
     public function add(Loader|Transformer $pipe) : Pipeline
@@ -60,18 +62,11 @@ public function pipes() : Pipes
 
     public function process(FlowContext $context) : \Generator
     {
-        foreach ($this->nextPipeline->setSource(new Extractor\PipelineExtractor($this->pipeline))->process($context) as $rows) {
+        foreach ($this->nextPipeline->process($context) as $rows) {
             yield $rows;
         }
     }
 
-    public function setSource(Extractor $extractor) : Pipeline
-    {
-        $this->pipeline->setSource($extractor);
-
-        return $this;
-    }
-
     public function source() : Extractor
     {
         return $this->pipeline->source();

src/core/etl/src/Flow/ETL/Pipeline/PartitioningPipeline.php

@@ -78,13 +78,6 @@ public function process(FlowContext $context) : \Generator
         )->extract($context);
     }
 
-    public function setSource(Extractor $extractor) : Pipeline
-    {
-        $this->pipeline->setSource($extractor);
-
-        return $this;
-    }
-
     public function source() : Extractor
     {
         return $this->pipeline->source();