iterate() : map an input Iterable to an output Iterable while applyin…

…g the pipeline to it

README.md

@@ -28,7 +28,7 @@ composer require camille-hdl/lazy-lists
 
 ## Documentation
 
-You can either use the functions directly on arrays or `Traversable`s
+You can use the functions directly on arrays or `Traversable`s
 
 ```php
 use LazyLists\map;
@@ -44,25 +44,49 @@ $result = flatten(1, $input);
 $result = take(10, $input);
 ```
 
-but the most interesting way of using LazyLists is to compose a "lazy pipeline".  
+but the most interesting way of using LazyLists is to use the composition functions : `pipe` or `iterate`.  
 Steps in the pipeline are executed sequentially *for each element* in the collection *in a single iteration*, unlike `array_map`, `array_filter` or other similar libraries.
 
 ```php
 use LazyLists\pipe;
+use LazyLists\iterate;
 use LazyLists\map;
 use LazyLists\filter;
 use LazyLists\reduce;
+use LazyLists\each;
 use LazyLists\flatten;
 use LazyLists\take;
+use LazyLists\until;
 
 $pipeline = pipe(
     flatten(1),
     filter($myPredicate),
     map($myTransformation),
+    each($mySideEffect),
     take(10),
     reduce($myAggregator, 0)
 );
+// returns an array
 $result = $pipeline($myArrayOrIterator);
+
+// returns an iterator
+$filterIterator = iterate(
+    filter($myPredicate),
+    until($myCondition)
+);
+foreach ($filterIterator($myArrayOrIterator) as $key => $value) {
+    echo "$key : $value";
+}
+
+// you can iterate over a reduction
+$reduceIterator = iterate(
+    reduce(function ($acc, $v) { return $acc + $v; }, 0),
+    until(function ($sum) { return $sum > 10; })
+);
+foreach ($reduceIterator([1, 5, 10]) as $reduction) {
+    echo $reduction;
+}
+// 1, 5
 ```
 
 ## Contributing

composer.json

@@ -37,6 +37,7 @@
             "src/isAssociativeArray.php",
             "src/map.php",
             "src/pipe.php",
+            "src/iterate.php",
             "src/filter.php",
             "src/reduce.php",
             "src/take.php",

src/LazyIterator.php

@@ -0,0 +1,146 @@
+<?php
+
+/**
+ * This file is part of the camille-hdl/lazy-lists library
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ *
+ * @copyright Copyright (c) Camille Hodoul <[email protected]>
+ * @license http://opensource.org/licenses/MIT MIT
+ */
+
+declare(strict_types=1);
+
+namespace LazyLists;
+
+/**
+ * Provides an Iterator interface to a LazyWorker
+ *
+ * This class can be instanciated by `LazyLists\iterate()`
+ * @see \LazyLists\iterate
+ */
+class LazyIterator extends LazyWorker implements \Iterator
+{
+    protected $key = 0;
+    protected $currentValueBuffer = [];
+    protected $currentValue = null;
+    protected $hasValue = false;
+    protected $canLoop = false;
+    /**
+     * @param array|\Iterator $subject
+     * @param array $transducers
+     */
+    public function __construct($subject, array $transducers)
+    {
+        parent::__construct($subject, $transducers);
+        $this->registerValueCallback(function ($value) {
+            $this->currentValueBuffer[] = $value;
+            $this->canLoop = false;
+        });
+    }
+
+    public function iteratorInitialization()
+    {
+        $this->key = 0;
+        $this->reset();
+        if ($this->iterator->valid()) {
+            $this->currentWorkingValue = $this->iterator->current();
+            $this->finalResultSoFar = $this->getLastTransducer()->getEmptyFinalResult();
+        }
+        $this->computeFirstValue();
+    }
+
+    /**
+     * @return void
+     */
+    protected function reset()
+    {
+        $this->initializeTransducers();
+        $this->resetTransducers();
+        $this->shouldCompleteEarly = false;
+        $this->completedEarly = false;
+        $this->iterator->rewind();
+    }
+
+    /**
+     * @return void
+     */
+    public function rewind()
+    {
+        $this->iteratorInitialization();
+    }
+
+    /**
+     * We need to read the first value in advance,
+     * basically, the iterator wrapped in $this->iterator is
+     * one step ahead
+     *
+     * @return void
+     */
+    protected function computeFirstValue()
+    {
+        if ($this->key !== 0) {
+            throw new \LogicException(
+                "Should not use computeFirstValue after first key"
+            );
+        }
+        $this->loopUntilNextValue();
+        $this->setCurrentValueFromBuffer();
+    }
+
+    protected function loopUntilNextValue()
+    {
+        $this->canLoop = true;
+        $this->loop();
+    }
+
+    protected function setCurrentValueFromBuffer()
+    {
+        if (\count($this->currentValueBuffer) > 0) {
+            $this->hasValue = true;
+            $this->currentValue = \array_shift($this->currentValueBuffer);
+        }
+    }
+
+    protected function computeCurrentValue()
+    {
+        $this->hasValue = false;
+        $this->key++;
+        if (\count($this->currentValueBuffer) <= 0) {
+            $this->loopUntilNextValue();
+        }
+        $this->setCurrentValueFromBuffer();
+    }
+
+    public function valid()
+    {
+        return $this->hasValue;
+    }
+
+    public function next()
+    {
+        $this->computeCurrentValue();
+    }
+
+    public function current()
+    {
+        return $this->currentValue;
+    }
+
+    public function key()
+    {
+        return $this->key;
+    }
+
+    /**
+     * @return void
+     */
+    protected function loop()
+    {
+        while ($this->canLoop && $this->iterator->valid() && !$this->completedEarly) {
+            $currentTransducer = $this->getCurrentTransducer();
+            $currentTransducer($this->currentWorkingValue);
+        }
+    }
+}

src/LazyWorker.php

@@ -40,7 +40,7 @@ class LazyWorker
     /**
      * @var int
      */
-    protected $currentTransducerIndex;
+    protected $currentTransducerIndex = 0;
     /**
      * @var mixed[]
      */
@@ -66,6 +66,14 @@ class LazyWorker
      */
     protected $completedEarly = false;
 
+    /**
+     * List of callables to be invoked whenever a new value
+     * is obtained
+     *
+     * @var array
+     */
+    protected $newValueCallbacks = [];
+
     /**
      * @param array|\Iterator $subject
      * @param array $transducers
@@ -86,10 +94,39 @@ public function __construct($subject, array $transducers)
      */
     public function __invoke()
     {
-        $this->initializeTransducers();
+        return $this->run();
+    }
+
+    /**
+     * $callback will be invoked whenever a new value is found:
+     * that is whenever a value is "outputted" by the pipeline.
+     * The $callback takes the value as single argument.
+     *
+     * @param callable $callback
+     * @return void
+     */
+    public function registerValueCallback(callable $callback)
+    {
+        $this->newValueCallbacks[] = $callback;
+    }
+
+    /**
+     * @return void
+     */
+    protected function reset()
+    {
+        $this->initializeTransducersAndLoop();
         $this->shouldCompleteEarly = false;
         $this->completedEarly = false;
         $this->iterator->rewind();
+    }
+
+    /**
+     * @return mixed
+     */
+    protected function run()
+    {
+        $this->reset();
         if (!$this->iterator->valid()) {
             return $this->finalResultSoFar;
         }
@@ -110,6 +147,19 @@ protected function loop()
         }
     }
 
+    /**
+     * Call callbacks whenever a new value is found
+     *
+     * @param mixed $value
+     * @return void
+     */
+    protected function onNewValue($value)
+    {
+        foreach ($this->newValueCallbacks as $callback) {
+            $callback($value);
+        }
+    }
+
     /**
      * During iteration, provides the next item to be
      * processed, either from the wrapped Iterator or
@@ -121,9 +171,10 @@ public function readNextItem()
     {
         $computedValuesInfo = $this->computedValuesToProcess();
         if (!\is_null($computedValuesInfo)) {
-            return $this->readComputedValueToProcessForIndex(
+            $computedValue = $this->readComputedValueToProcessForIndex(
                 $computedValuesInfo["index"]
             );
+            return $computedValue;
         }
         $this->iterator->next();
         if ($this->iterator->valid()) {
@@ -271,17 +322,15 @@ protected function computedValuesToProcess()
     }
 
     /**
-     * Starts a new loop in the transducers,
-     * and reads the next value to be processed.
-     *
-     * If the value is provided from a transducers (as opposed to
+     * Starts a new loop in the transducers.
+     * If there are values provided from transducers (as opposed to
      * the wrapped iterator), we don't start the loop from the
      * beginning but only from the Transducers downstream from the one
      * who provided the value.
      *
      * @return void
      */
-    protected function resetLoop()
+    protected function resetTransducers()
     {
         $computedValuesToProcess = $this->computedValuesToProcess();
         if (\is_null($computedValuesToProcess)) {
@@ -290,11 +339,27 @@ protected function resetLoop()
         } else {
             $this->currentTransducerIndex = $computedValuesToProcess["index"];
         }
+    }
+
+    /**
+     * Starts a new loop in the transducers,
+     * and reads the next value to be processed.
+     *
+     * @return void
+     */
+    protected function resetLoop()
+    {
+        $this->resetTransducers();
         $this->currentWorkingValue = $this->readNextItem();
     }
 
     /**
-     * Undocumented function
+     * This reads and returns all "future" values :
+     * the values that a transducer upstream made available to be processed.
+     * This is used in case we need to finish processing while some values
+     * haven't been processed by another transducer and thus incorporated in the
+     * final result.
+     * Typically: when `flatten()` is the last transducer in the pipeline.
      *
      * @param integer $fromIndex
      * @return array
@@ -331,6 +396,7 @@ protected function readAllFutureComputedValues(int $fromIndex): array
     protected function updateFinalResult()
     {
         $lastTransducer = $this->getLastTransducer();
+        $this->onNewValue($this->currentWorkingValue);
         $this->finalResultSoFar = $lastTransducer->computeFinalResult(
             $this->finalResultSoFar,
             $this->currentWorkingValue
@@ -342,6 +408,7 @@ protected function updateFinalResult()
         foreach (
             $this->readAllFutureComputedValues($this->currentTransducerIndex) as $value
         ) {
+            $this->onNewValue($value);
             $this->finalResultSoFar = $lastTransducer->computeFinalResult(
                 $this->finalResultSoFar,
                 $value
@@ -352,9 +419,17 @@ protected function updateFinalResult()
     /**
      * @return void
      */
-    protected function initializeTransducers()
+    protected function initializeTransducersAndLoop()
     {
         $this->resetLoop();
+        $this->initializeTransducers();
+    }
+
+    /**
+     * @return void
+     */
+    protected function initializeTransducers()
+    {
         foreach ($this->transducers as $transducer) {
             $transducer->initialize($this);
         }