Merge pull request #132 from mindaugasbarysas/pipeline-leak

Fixes for pipeline leaks, introduces breaking changes.

EventListener/AbstractConsumeEventListener.php

@@ -25,7 +25,7 @@ abstract class AbstractConsumeEventListener
      */
     public function onConsume(ItemPipelineEvent $event)
     {
-        if ($event->getSkipException()) {
+        if ($event->getItemSkip()) {
             $this->skip($event);
         } else {
             $this->consume($event);

Pipeline/Event/ItemPipelineEvent.php

@@ -11,7 +11,7 @@
 
 namespace ONGR\ConnectionsBundle\Pipeline\Event;
 
-use ONGR\ConnectionsBundle\Pipeline\ItemSkipException;
+use ONGR\ConnectionsBundle\Pipeline\ItemSkip;
 use Symfony\Component\EventDispatcher\Event;
 
 /**
@@ -32,9 +32,9 @@ class ItemPipelineEvent extends Event
     private $output;
 
     /**
-     * @var ItemSkipException
+     * @var ItemSkip
      */
-    private $skipException;
+    private $itemSkip;
 
     /**
      * @param mixed $item
@@ -77,21 +77,21 @@ public function setOutput($output)
     }
 
     /**
-     * @return ItemSkipException
+     * @return ItemSkip
      */
-    public function getSkipException()
+    public function getItemSkip()
     {
-        return $this->skipException;
+        return $this->itemSkip;
     }
 
     /**
-     * @param ItemSkipException $skipException
+     * @param ItemSkip $itemSkip
      *
      * @return $this
      */
-    public function setSkipException(ItemSkipException $skipException)
+    public function setItemSkip(ItemSkip $itemSkip)
     {
-        $this->skipException = $skipException;
+        $this->itemSkip = $itemSkip;
 
         return $this;
     }

Pipeline/ItemSkip.php

@@ -0,0 +1,39 @@
+<?php
+
+/*
+ * This file is part of the ONGR package.
+ *
+ * (c) NFQ Technologies UAB <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace ONGR\ConnectionsBundle\Pipeline;
+
+/**
+ * Class for skipping items.
+ */
+class ItemSkip
+{
+    /**
+     * @var string Reason for skipping.
+     */
+    private $reason;
+
+    /**
+     * @return string
+     */
+    public function getReason()
+    {
+        return $this->reason;
+    }
+
+    /**
+     * @param string $reason
+     */
+    public function setReason($reason)
+    {
+        $this->reason = $reason;
+    }
+}

Pipeline/ItemSkipper.php

@@ -0,0 +1,34 @@
+<?php
+
+/*
+ * This file is part of the ONGR package.
+ *
+ * (c) NFQ Technologies UAB <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace ONGR\ConnectionsBundle\Pipeline;
+
+use ONGR\ConnectionsBundle\Pipeline\Event\ItemPipelineEvent;
+
+/**
+ * Provides a static method for skipping unwanted items.
+ */
+class ItemSkipper
+{
+    /**
+     * Skips item, marks event as skipped with a reason.
+     *
+     * @param ItemPipelineEvent $event
+     * @param string            $reason
+     */
+    public static function skip(ItemPipelineEvent $event, $reason = '')
+    {
+        $itemSkip = new ItemSkip();
+        $itemSkip->setReason($reason);
+        $event->setItemSkip($itemSkip);
+        $event->stopPropagation();
+    }
+}

Pipeline/Pipeline.php

@@ -71,14 +71,10 @@ public function start()
                 $itemEvent = new ItemPipelineEvent($item);
                 $itemEvent->setContext($this->getContext());
 
-                try {
-                    $dispatcher->dispatch(
-                        $this->getEventName(self::EVENT_SUFFIX_MODIFY),
-                        $itemEvent
-                    );
-                } catch (ItemSkipException $itemSkipException) {
-                    $itemEvent->setSkipException($itemSkipException);
-                }
+                $dispatcher->dispatch(
+                    $this->getEventName(self::EVENT_SUFFIX_MODIFY),
+                    $itemEvent
+                );
 
                 $dispatcher->dispatch(
                     $this->getEventName(self::EVENT_SUFFIX_CONSUME),

Resources/doc/Pipeline/pipeline.rst

@@ -31,8 +31,10 @@ Example:
 
 Item skipping
 -------------
-If item for some reason should be skipped without stopping pipeline, ItemSkipException can be used.
+If item for some reason should be skipped without stopping pipeline, ItemSkipper static class 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.
+When modifier invokes ``ItemSkipper::skip`` method, it sets ``ItemSkip`` object in ``ItemPipelineEvent`` with a reason
+for skipping (optional).
+
+If ``AbstractConsumeEventListener`` is used and ``ItemSkip`` is set, ``skip`` method will be called.
 Otherwise ``consume`` will be invoked.

Tests/Unit/Pipeline/PipelineTest.php

@@ -13,7 +13,7 @@
 
 use ONGR\ConnectionsBundle\Pipeline\Event\ItemPipelineEvent;
 use ONGR\ConnectionsBundle\Pipeline\Event\SourcePipelineEvent;
-use ONGR\ConnectionsBundle\Pipeline\ItemSkipException;
+use ONGR\ConnectionsBundle\Pipeline\ItemSkipper;
 use ONGR\ConnectionsBundle\Pipeline\PipelineFactory;
 use Symfony\Component\EventDispatcher\EventDispatcher;
 
@@ -82,13 +82,11 @@ public function testPipeline($data, $expectedConsumedItems, $expectedSkippedItem
      * OnModify.
      *
      * @param ItemPipelineEvent $event
-     *
-     * @throws ItemSkipException
      */
     public function onModify(ItemPipelineEvent $event)
     {
         if ($event->getItem() == 'skip') {
-            throw new ItemSkipException();
+            ItemSkipper::skip($event, 'Test reason for skip');
         }
     }
 }

Tests/Unit/Pipeline/PipelineTestConsumer.php

@@ -42,7 +42,9 @@ public function consume(ItemPipelineEvent $event)
      */
     public function skip(ItemPipelineEvent $event)
     {
-        $this->skipCalled++;
+        if ($event->getItemSkip()->getReason() === 'Test reason for skip') {
+            $this->skipCalled++;
+        }
     }
 
     /**