Change pipeline implementation

Author: leomarquine

src/Pipeline.php

@@ -2,239 +2,225 @@
 
 namespace Marquine\Etl;
 
-use IteratorAggregate;
+use Iterator;
+use Marquine\Etl\Loaders\Loader;
+use Marquine\Etl\Extractors\Extractor;
+use Marquine\Etl\Transformers\Transformer;
 
-class Pipeline
+class Pipeline implements Iterator
 {
     /**
      * The pipeline data flow.
      *
-     * @var \IteratorAggregate
+     * @var \Generator
      */
     protected $flow;
 
     /**
-     * The array of tasks.
+     * The maximum number of rows.
      *
-     * @var array
+     * @var int
      */
-    protected $tasks = [];
+    protected $limit;
 
     /**
-     * Current row.
+     * The number of rows to skip.
      *
      * @var int
      */
-    protected $current = 0;
+    protected $skip;
 
     /**
-     * Total rows.
+     * The iteration key.
      *
      * @var int
      */
-    protected $total = 0;
+    protected $key;
 
     /**
-     * Maximum number of rows.
+     * The current iteration row.
      *
-     * @var int
+     * @var \Marquine\Etl\Row
      */
-    protected $limit;
+    protected $current;
 
     /**
-     * Number of rows to skip.
+     * The etl extractor.
      *
-     * @var int
+     * @var \Marquine\Etl\Extractors\Extractor
      */
-    protected $skip;
+    protected $extractor;
 
     /**
-     * Pre execution callbacks.
+     * The array of steps for the pipeline.
      *
      * @var array
      */
-    protected $preExecutionCallbacks = [];
+    protected $steps = [];
 
     /**
-     * Post execution callbacks.
+     * Set the pipeline extractor.
      *
-     * @var array
+     * @param  \Marquine\Etl\Extractors\Extractor  $extractor
+     * @return void
      */
-    protected $postExecutionCallbacks = [];
+    public function extractor(Extractor $extractor)
+    {
+        $this->extractor = $extractor;
+    }
 
     /**
-     * Set the pipeline flow.
+     * Add a step to the pipeline.
      *
-     * @param  \IteratorAggregate  $flow
+     * @param  \Marquine\EtlStep  $step
      * @return void
      */
-    public function flow(IteratorAggregate $flow)
+    public function pipe(Step $step)
     {
-        $this->flow = $flow;
+        $this->steps[] = $step;
     }
 
     /**
-     * Pipe a task.
+     * Set the row limit.
      *
-     * @param  callable  $task
-     * @return $this
+     * @param  int  $limit
+     * @return void
      */
-    public function pipe(callable $task)
+    public function limit($limit)
     {
-        $this->tasks[] = $task;
-
-        return $this;
+        $this->limit = $limit;
     }
 
     /**
-     * Get the pipeline data generator.
+     * Set the number of rows to skip.
      *
-     * @return \Generator
+     * @param  int  $skip
+     * @return void
      */
-    public function get()
+    public function skip($skip)
     {
-        $this->total = $this->getRowsCount();
-
-        foreach ($this->preExecutionCallbacks as $callback) {
-            $callback();
-        }
-
-        foreach ($this->flow as $index => $row) {
-            if ($this->skip && $index < $this->skip) {
-                continue;
-            }
-
-            if ($this->limit && $this->current == $this->limit) {
-                break;
-            }
-
-            $this->current++;
-
-            $row = $this->runTasks($row);
-
-            if (!empty($row)) {
-                yield $row;
-            }
-        }
-
-        foreach ($this->postExecutionCallbacks as $callback) {
-            $callback();
-        }
+        $this->skip = $skip;
     }
 
     /**
-     * Get a sample row of the flow.
+     * Get the current element.
      *
-     * @return array
+     * @return void
      */
-    public function sample()
+    public function current()
     {
-        return $this->flow->getIterator()->current();
+        return $this->current->toArray();
     }
 
     /**
-     * Run tasks for the given row.
+     * Move forward to next element.
      *
-     * @param  array  $row
-     * @return array
+     * @return void
      */
-    protected function runTasks($row)
+    public function next()
     {
-        foreach ($this->tasks as $task) {
-            $row = $task($row);
-
-            if (empty($row)) {
-                break;
-            }
-        }
-
-        return $row;
+        $this->key++;
+        $this->flow->next();
     }
 
     /**
-     * Get the total rows count.
+     * Get the key of the current element.
      *
      * @return int
      */
-    protected function getRowsCount()
+    public function key()
     {
-        $count = iterator_count($this->flow);
-
-        if ($this->skip) {
-            $count -= $this->skip;
-        }
-
-        if ($this->limit && $count > $this->limit) {
-            $count = $this->limit;
-        }
-
-        return $count;
+        return $this->key;
     }
 
     /**
-     * Get the metadata.
+     * Checks if current position is valid.
      *
-     * @return array
+     * @return bool
      */
-    public function metadata($attribute = null)
+    public function valid()
     {
-        $metadata = [
-            'current' => $this->current,
-            'total' => $this->total,
-        ];
+        if (! $this->flow->valid() || $this->limitReached()) {
+            $this->finalize();
+
+            return false;
+        }
+
+        $this->current = $this->flow->current();
+
+
+        foreach ($this->steps as $step) {
+            if ($this->current->discarded()) {
+                $this->key--;
+                $this->next();
+
+                return $this->valid();
+            }
+
+            if ($step instanceof Transformer) {
+                $step->transform($this->current);
+            }
+
+            if ($step instanceof Loader) {
+                $step->load($this->current);
+            }
+        }
 
-        return $metadata[$attribute] ?? (object) $metadata;
+        return true;
     }
 
     /**
-     * Set the maximum number of rows.
+     * Rewind the Iterator to the first element.
      *
-     * @param  int  $value
-     * @return $this
+     * @return void
      */
-    public function limit($value)
+    public function rewind()
     {
-        $this->limit = $value;
+        $this->initialize();
+
+        $this->key = 0;
+        $this->flow = $this->extractor->extract();
 
-        return $this;
+        while ($this->flow->key() < $this->skip && $this->flow->valid()) {
+            $this->flow->next();
+        }
     }
 
     /**
-     * Set the number of rows to skip.
+     * Check if the row limit was reached.
      *
-     * @param  int  $value
-     * @return $this
+     * @return bool
      */
-    public function skip($value)
+    protected function limitReached()
     {
-        $this->skip = $value;
-
-        return $this;
+        return $this->limit && $this->key() === $this->limit;
     }
 
     /**
-     * Register pre execution callback.
+     * Initialize the steps.
      *
-     * @param  callable  $callback
-     * @return $this
+     * @return void
      */
-    public function before(callable $callback)
+    protected function initialize()
     {
-        $this->preExecutionCallbacks[] = $callback;
+        $this->extractor->initialize();
 
-        return $this;
+        foreach ($this->steps as $step) {
+            $step->initialize();
+        }
     }
 
     /**
-     * Register post execution callback.
+     * Finalize the steps.
      *
-     * @param  callable  $callback
-     * @return $this
+     * @return void
      */
-    public function after(callable $callback)
+    protected function finalize()
     {
-        $this->postExecutionCallbacks[] = $callback;
+        $this->extractor->finalize();
 
-        return $this;
+        foreach ($this->steps as $step) {
+            $step->finalize();
+        }
     }
 }