Fixes thephpleague#38, Implement Fork-Join with builder support

spec/NewBuilderSpec.php

@@ -0,0 +1,55 @@
+<?php
+/**
+ * Created by PhpStorm.
+ * User: juzerali
+ * Date: 09/01/18
+ * Time: 8:20 PM
+ */
+
+namespace spec\League\Pipeline;
+
+
+use League\Pipeline\PipelineBuilder;
+use PhpSpec\Exception\Exception;
+use PhpSpec\ObjectBehavior;
+
+class NewBuilderSpec extends ObjectBehavior
+{
+    public function it_should_pass()
+    {
+        $pipeline = (new PipelineBuilder())
+            ->pipe(function($payload) {return $payload * 2;})
+            ->fork(function($payload) {
+                if($payload == 0) return '0';
+                if($payload > 0) return '+';
+                if($payload < 0) return '-';
+                return false;
+            })
+            ->disjoin('0')
+            ->pipe(function() {return INF;})
+            ->disjoin('+')
+            ->pipe(function($payload) {return sqrt($payload);})
+            ->pipe(function($payload) {return $payload / 2;})
+            ->disjoin('-', function() {return NAN;})
+            ->join()
+            ->pipe(function ($payload) {return "&" . $payload;})
+            ->build()
+        ;
+
+        if(($result = $pipeline->process(0)) != '&INF')
+        {
+            throw new Exception('Should be &INF but was ' . $result);
+        }
+
+        if(($result = $pipeline->process(32)) != '&4')
+        {
+            throw new Exception('Should be &4 but was ' . $result);
+        }
+
+        if($pipeline->process(-1) != '&NAN')
+        {
+            throw new Exception('Should be NAN');
+        }
+    }
+
+}

src/BuilderInterface.php

@@ -0,0 +1,15 @@
+<?php
+/**
+ * Created by PhpStorm.
+ * User: juzerali
+ * Date: 09/01/18
+ * Time: 7:47 PM
+ */
+
+namespace League\Pipeline;
+
+
+interface BuilderInterface
+{
+    public function build();
+}

src/DisjointPipelineBuilder.php

@@ -0,0 +1,31 @@
+<?php
+/**
+ * Created by PhpStorm.
+ * User: juzerali
+ * Date: 09/01/18
+ * Time: 6:52 PM
+ */
+
+namespace League\Pipeline;
+
+
+class DisjointPipelineBuilder extends PipelineBuilder implements DisjointPipelineBuilderInterface
+{
+    private $parent;
+
+    public function __construct(ForkBuilder $parent, callable $stage = null)
+    {
+        if($stage != null) $this->stages[] = $stage;
+        $this->parent = $parent;
+    }
+
+    public function disjoin(string $tag, callable $stage = null)
+    {
+        return $this->parent->disjoin($tag, $stage);
+    }
+
+    public function join()
+    {
+        return $this->parent->join();
+    }
+}

src/DisjointPipelineBuilderInterface.php

@@ -0,0 +1,15 @@
+<?php
+/**
+ * Created by PhpStorm.
+ * User: juzerali
+ * Date: 09/01/18
+ * Time: 6:51 PM
+ */
+
+namespace League\Pipeline;
+
+
+interface DisjointPipelineBuilderInterface extends PipelineBuilderInterface, ForkBuilderInterface
+{
+    public function join();
+}

src/Fork.php

@@ -22,12 +22,18 @@ class Fork implements ForkInterface
 
     /**
      * Fork constructor.
-     *
      * @param callable|null $resolver
+     * @param array $forks
      */
-    public function __construct(callable $resolver = null)
+    public function __construct(callable $resolver = null, $forks = [])
     {
         $this->resolver = $resolver;
+        $this->forks = $forks;
+    }
+
+    public function setForks($forks)
+    {
+        $this->forks = $forks;
     }
 
     public function pipeline(Pipeline $pipeline)

src/ForkBuilder.php

@@ -0,0 +1,63 @@
+<?php
+/**
+ * Created by PhpStorm.
+ * User: juzerali
+ * Date: 09/01/18
+ * Time: 7:35 PM
+ */
+
+namespace League\Pipeline;
+
+
+class ForkBuilder implements ForkBuilderInterface
+{
+    /**
+     * @var PipelineBuilder
+     */
+    private $parent;
+
+
+    private $forks = [];
+
+    private $resolver;
+
+    /**
+     * @var BuilderInterface
+     */
+    private $currentBuilder;
+    private $currentTag;
+
+    public function __construct(callable $resolver, PipelineBuilder $parent)
+    {
+        $this->resolver = $resolver;
+        $this->parent = $parent;
+    }
+
+    public function disjoin(string $tag, callable $stage = null)
+    {
+        $this->buildPrevious();
+        $this->currentTag = $tag;
+        $this->currentBuilder = new DisjointPipelineBuilder($this, $stage);
+        return $this->currentBuilder;
+    }
+
+    public function join()
+    {
+        $this->buildPrevious();
+        return $this->parent;
+    }
+
+    private function buildPrevious()
+    {
+        if($this->currentTag == null) return;
+        $this->forks[$this->currentTag] = $this->currentBuilder->build();
+
+        $this->currentTag = null;
+        $this->currentBuilder = null;
+    }
+
+    public function build()
+    {
+        return new Fork($this->resolver, $this->forks);
+    }
+}

src/ForkBuilderInterface.php

@@ -0,0 +1,15 @@
+<?php
+/**
+ * Created by PhpStorm.
+ * User: juzerali
+ * Date: 09/01/18
+ * Time: 7:28 PM
+ */
+
+namespace League\Pipeline;
+
+
+interface ForkBuilderInterface extends BuilderInterface
+{
+    public function disjoin(string $tag, callable $stage = null);
+}

src/ForkedPipelineBuilderInterface.php

@@ -0,0 +1,15 @@
+<?php
+/**
+ * Created by PhpStorm.
+ * User: juzerali
+ * Date: 09/01/18
+ * Time: 7:29 PM
+ */
+
+namespace League\Pipeline;
+
+
+interface ForkedPipelineBuilderInterface extends ForkBuilderInterface
+{
+    public function join();
+}

src/PipelineBuilder.php

@@ -2,12 +2,24 @@
 
 namespace League\Pipeline;
 
+use Illuminate\Database\Schema\Builder;
+
 class PipelineBuilder implements PipelineBuilderInterface
 {
     /**
      * @var callable[]
      */
-    private $stages = [];
+    protected $stages = [];
+
+    /**
+     * @var ProcessorInterface
+     */
+    private $processr;
+
+    /**
+     * @var BuilderInterface
+     */
+    private $last;
 
     /**
      * Add an stage.
@@ -16,22 +28,41 @@ class PipelineBuilder implements PipelineBuilderInterface
      *
      * @return $this
      */
-    public function add(callable $stage)
+    public function pipe(callable $stage)
     {
+        $this->buildLast();
         $this->stages[] = $stage;
 
         return $this;
     }
 
+    public function fork(callable $resolver)
+    {
+        $this->buildLast();
+        $this->last = new ForkBuilder($resolver, $this);
+        return $this->last;
+    }
+
+    public function processor(ProcessorInterface $processor)
+    {
+        $this->processr = $processor;
+    }
+
+    private function buildLast()
+    {
+        if($this->last == null) return;
+        $this->stages[] = $this->last->build();
+        $this->last = null;
+    }
+
     /**
      * Build a new Pipeline object
      *
-     * @param  ProcessorInterface|null $processor
-     *
-     * @return PipelineInterface
+     * @return Pipeline
      */
-    public function build(ProcessorInterface $processor = null)
+    public function build()
     {
-        return new Pipeline($this->stages, $processor);
+        $this->buildLast();
+        return new Pipeline($this->stages, $this->processr);
     }
 }

src/PipelineBuilderInterface.php

@@ -1,23 +1,30 @@
 <?php
 namespace League\Pipeline;
 
-interface PipelineBuilderInterface
+interface PipelineBuilderInterface extends BuilderInterface
 {
     /**
-     * Add an stage.
+     * Add a stage.
      *
      * @param callable $stage
      *
      * @return $this
      */
-    public function add(callable $stage);
+    public function pipe(callable $stage);
+
+    /**
+     * Forks the pipeline
+     *
+     * @param callable $resolver
+     * @return mixed
+     */
+    public function fork(callable $resolver);
 
     /**
      * Build a new Pipeline object
      *
-     * @param  ProcessorInterface|null $processor
      *
      * @return PipelineInterface
      */
-    public function build(ProcessorInterface $processor = null);
+    public function build();
 }