Initial implementation of processing

src/main/php/util/data/Processing.class.php

@@ -0,0 +1,95 @@
+<?php namespace util\data;
+
+/**
+ * Processing
+ *
+ * @test  xp://util.data.unittest.ProcessingTest
+ */
+class Processing extends \lang\Object implements \Iterator {
+  private $it, $func;
+  private $defer= [];
+
+  /**
+   * Creates a new Processor instance
+   *
+   * @param  php.Mapper $it
+   * @param  php.Closure $func
+   */
+  public function __construct(\Iterator $it, \Closure $func) {
+    $this->it= $it;
+    $this->func= $func;
+  }
+
+  private function process($key, $value) {
+    $this->key= $key;
+    $result= $this->func->__invoke($this, $value);
+    if (null === $this->key) {
+      return false;
+    } else {
+      $this->current= null === $result ? $value : $result;
+      return true;
+    }
+  }
+
+  public function defer($value) {
+    $key= $this->key;
+    $this->defer[]= function() use($key, $value) {
+      $this->key= $key;
+      $this->current= $value;
+      return true;
+    };
+    $this->key= null;
+  }
+
+  public function drop($value) {
+    $this->key= null;
+  }
+
+  public function retry($value, $closure= null) {
+    $key= $this->key;
+    $this->defer[]= function() use($key, $value, $closure) {
+      $closure && $closure();
+      return $this->process($key, $value);
+    };
+    $this->key= null;
+  }
+
+  /** @return void */
+  private function forward() {
+    while ($this->it->valid()) {
+      if ($this->process($this->it->key(), $this->it->current())) {
+        $this->valid= true;
+        return;
+      }
+
+      $this->it->next();
+    }
+
+    if ($this->valid= !empty($this->defer)) {
+      do {
+        $handled= array_shift($this->defer);
+      } while (!$handled());
+    }
+  }
+
+  /** @return void */
+  public function rewind() {
+    $this->it->rewind();
+    $this->forward();
+  }
+
+  /** @return void */
+  public function next() {
+    $this->it->next();
+    $this->forward();
+  }
+
+  /** @return var */
+  public function current() { return $this->current; }
+
+  /** @return var */
+  public function key() { return $this->key; }
+
+  /** @return bool */
+  public function valid() { return $this->valid; }
+}

src/main/php/util/data/Sequence.class.php

@@ -378,6 +378,17 @@ public function map($function) {
     return new self($m);
   }
 
+  /**
+   * Returns a new stream which calls a processing function for each element.
+   *
+   * @param  function(var): var $function
+   * @return self
+   * @throws lang.IllegalArgumentException
+   */
+  public function process($function) {
+    return new self(new Processing($this->getIterator(), Functions::$APPLY->newInstance($function)));
+  }
+
   /**
    * Returns a new stream which flattens, mapping the given function to each
    * element.

src/test/php/util/data/unittest/ProcessingTest.class.php

@@ -0,0 +1,50 @@
+<?php namespace util\data\unittest;
+
+use util\data\Sequence;
+use util\data\Processing;
+use lang\IndexOutOfBoundsException;
+
+class ProcessingTest extends \unittest\TestCase {
+
+  #[@test]
+  public function defer() {
+    $result= Sequence::of([1, 2, 3, 4])
+      ->process(function($processing, $i) {
+        if (0 === $i % 2) $processing->defer($i);
+      })
+      ->map(function($i) { return 2 * $i; })
+    ;
+
+    $this->assertEquals([2, 6, 4, 8], $result->toArray());
+  }
+
+  #[@test]
+  public function drop() {
+    $result= Sequence::of([1, 2, 3, 4])
+      ->process(function($processing, $i) {
+        if (0 === $i % 2) $processing->drop($i);
+      })
+      ->map(function($i) { return 2 * $i; })
+    ;
+
+    $this->assertEquals([2, 6], $result->toArray());
+  }
+
+  #[@test]
+  public function retry() {
+    $result= Sequence::of(['hello', '  hi'])
+      ->process(function($processing, $index) {
+        static $map= ['hello' => 1, 'hi' => 2];
+
+        try {
+          return $map[$index];
+        } catch (IndexOutOfBoundsException $t) {
+          $processing->retry(substr($index, 1));
+        }
+      })
+      ->map(function($i) { return 2 * $i; })
+    ;
+
+    $this->assertEquals([2, 4], $result->toArray());
+  }
+}