Fixed Tautrimas comments and added basic pipeline documentation.

EventListener/AbstractConsumeEventListener.php

@@ -14,7 +14,7 @@
 use ONGR\ConnectionsBundle\Pipeline\Event\ItemPipelineEvent;
 
 /**
- * AbstractConsumeEventListener class.
+ * Provides functionality for skipping pipeline items.
  */
 abstract class AbstractConsumeEventListener
 {
@@ -25,7 +25,7 @@ abstract class AbstractConsumeEventListener
      */
     public function onConsume(ItemPipelineEvent $event)
     {
-        if ($event->getItemSkipException()) {
+        if ($event->getSkipException()) {
             $this->skip($event);
         } else {
             $this->consume($event);

Pipeline/Event/ItemPipelineEvent.php

@@ -34,7 +34,7 @@ class ItemPipelineEvent extends Event
     /**
      * @var ItemSkipException
      */
-    private $itemSkipException;
+    private $skipException;
 
     /**
      * @param mixed $item
@@ -79,19 +79,19 @@ public function setOutput($output)
     /**
      * @return ItemSkipException
      */
-    public function getItemSkipException()
+    public function getSkipException()
     {
-        return $this->itemSkipException;
+        return $this->skipException;
     }
 
     /**
-     * @param ItemSkipException $itemSkipException
+     * @param ItemSkipException $skipException
      *
      * @return $this
      */
-    public function setItemSkipException(ItemSkipException $itemSkipException)
+    public function setSkipException(ItemSkipException $skipException)
     {
-        $this->itemSkipException = $itemSkipException;
+        $this->skipException = $skipException;
 
         return $this;
     }

Pipeline/ItemSkipException.php

@@ -12,7 +12,12 @@
 namespace ONGR\ConnectionsBundle\Pipeline;
 
 /**
- * Skip Item Exception.
+ * Exception for skipping items.
+ *
+ * This exception (or exception derived from this)
+ * should be thrown inside pipeline modifier
+ * to indicate that this item should be skipped.
+ * Thrown exception will be available for consumers.
  */
 class ItemSkipException extends \Exception
 {

Pipeline/Pipeline.php

@@ -75,7 +75,7 @@ public function start()
                         $itemEvent
                     );
                 } catch (ItemSkipException $itemSkipException) {
-                    $itemEvent->setItemSkipException($itemSkipException);
+                    $itemEvent->setSkipException($itemSkipException);
                 }
 
                 $dispatcher->dispatch(

Resources/doc/pipeline.rst

@@ -0,0 +1,38 @@
+Pipeline
+========
+
+Pipeline is used to process data with 5 events:
+
+ - source
+ - start
+ - modify
+ - consume
+ - finish
+
+Pipeline starts with source event which provides all data which should be processed.
+Then start event is fired to indicate that items are about to come,
+after that pipeline iterates through all items from all sources, calling modify and consume events
+with each item inside ``ItemPipelineEvent``. After the iterations are finished, finish event is fired
+to notify that no more items will follow.
+
+Pipeline can have any number of listeners for each event but for functioning pipeline
+at least one source and consume listener should be provided.
+
+Listeners should listen to ongr.pipeline.<PipelineName>.<Target>.<Event> events.
+Example:
+
+.. code-block:: yaml
+
+    test.import.modifier:
+        class: %ongr_connections.import.modifier.class%
+        tags:
+            - { name: kernel.event_listener, event: ongr.pipeline.import.default.modify, method: onModify }
+..
+
+Item skipping
+-------------
+If item for some reason should be skipped without stopping pipeline, ItemSkipException can be used.
+
+When modifier throws ``ItemSkipException`` pipeline catches it and sets skipException in ``ItemPipelineEvent``.
+If ``AbstractConsumeEventListener`` is used and ``skipException`` exception is set, ``skip`` method will be called.
+Otherwise ``consume`` will be invoked.

Tests/Unit/Pipeline/PipelineTest.php

@@ -17,11 +17,30 @@
 use ONGR\ConnectionsBundle\Pipeline\PipelineFactory;
 use Symfony\Component\EventDispatcher\EventDispatcher;
 
-/**
- * PipelineTest class.
- */
 class PipelineTest extends \PHPUnit_Framework_TestCase
 {
+    /**
+     * Pipeline data provider.
+     *
+     * @return array
+     */
+    public function pipelineData()
+    {
+        // Modifier skips items which are equal to 'skip', anything else gets consumed.
+        return [
+            // Case #0: No data. Results should be: 0 consumed and skipped items.
+            [[], 0, 0],
+            // Case #1: All data should be consumed, so 1 consume and 0 skips.
+            [['consume'], 1, 0],
+            // Case #2: All data should be skipped, so 0 consumes and 1 skip.
+            [['skip'], 0, 1],
+            // Case #3: Data with consume and skips. 1 consume and 2 skips.
+            [['skip', 'consume', 'skip'], 1, 2],
+            // Case #4: Data with consumes and skip. 2 consumes and 1 skip.
+            [['consume', 'skip', 'consume'], 2, 1],
+        ];
+    }
+
     /**
      * Tests pipeline.
      *
@@ -70,20 +89,4 @@ public function onModify(ItemPipelineEvent $event)
             throw new ItemSkipException();
         }
     }
-
-    /**
-     * Pipeline data provider.
-     *
-     * @return array
-     */
-    public function pipelineData()
-    {
-        return [
-            [[], 0, 0],
-            [['consume'], 1, 0],
-            [['skip'], 0, 1],
-            [['skip', 'consume', 'skip'], 1, 2],
-            [['consume', 'skip', 'consume'], 2, 1],
-        ];
-    }
 }

Tests/Unit/Pipeline/PipelineTestConsumer.php

@@ -34,15 +34,15 @@ class PipelineTestConsumer extends AbstractConsumeEventListener
      */
     public function consume(ItemPipelineEvent $event)
     {
-        ++$this->consumeCalled;
+        $this->consumeCalled++;
     }
 
     /**
      * {@inheritdoc}
      */
     public function skip(ItemPipelineEvent $event)
     {
-        ++$this->skipCalled;
+        $this->skipCalled++;
     }
 
     /**