Add EventDispatcher to producer

spec/Connector/Writer/RabbitMqProducerSpec.php

@@ -7,26 +7,26 @@
 use Prophecy\Argument;
 use Sylake\AkeneoProducerBundle\Connector\Writer\RabbitMqProduct;
 use PhpSpec\ObjectBehavior;
+use Sylake\AkeneoProducerBundle\Event\MessageEvent;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
 final class RabbitMqProducerSpec extends ObjectBehavior
 {
-    function let(ProducerInterface $producer)
+    function let(ProducerInterface $producer, EventDispatcherInterface $eventDispatcher)
     {
-        $this->beConstructedWith($producer, 'message_type');
+        $this->beConstructedWith($producer, 'message_type', $eventDispatcher);
     }
 
     function it_is_an_akeneo_writer()
     {
         $this->shouldImplement(ItemWriterInterface::class);
     }
 
-    function it_publishes_product_created_events(ProducerInterface $producer)
+    function it_publishes_product_created_events(ProducerInterface $producer, EventDispatcherInterface $eventDispatcher)
     {
-        $this->beConstructedWith($producer, 'message_type');
+        $this->beConstructedWith($producer, 'message_type', $eventDispatcher);
 
-        $producer->publish(json_encode([
-            'type' => 'message_type',
-            'payload' => [
+        $item = [
                 'identifier' => 'AKNTS_BPXS',
                 'categories' => ['goodies', 'tshirts'],
                 'values' => [
@@ -59,111 +59,128 @@ function it_publishes_product_created_events(ProducerInterface $producer)
                         'data' => [['amount' => 20, 'currency' => 'USD']],
                     ],
                 ],
-            ],
+            ];
+
+
+        $eventDispatcher
+            ->dispatch('akeneo_producer_message.post_publish', new MessageEvent('message_type', $item))
+            ->shouldBeCalled();
+
+        $producer->publish(json_encode([
+            'type' => 'message_type',
+            'payload' => $item,
             'recordedOn' => (new \DateTime())->format('Y-m-d H:i:s'),
         ]))->shouldBeCalled();
 
         $this->write([
-            [
-                'identifier' => 'AKNTS_BPXS',
-                'categories' => ['goodies', 'tshirts'],
-                'values' => [
-                    'sku' => [['locale' => null, 'scope' => null]],
-                    'clothing_size' => [['locale' => null, 'scope' => null, 'data' => 'xs']],
-                    'description' => [['locale' => null, 'scope' => null, 'data' => 'Akeneo T-Shirt']],
-                    'tshirt_materials' => [['locale' => null, 'scope' => null, 'data' => 'cotton']],
-                    'tshirt_style' => [['locale' => null, 'scope' => null, 'data' => ['crewneck', 'short_sleeve']]],
-                ],
-                'created' => '2017-05-04T12:58:07+00:00',
-                'associations' => [
-                    'SUBSTITUTION' => [
-                        'groups' => [],
-                        'products' => [
-                            'AKNTS_WPXS',
-                            'AKNTS_PBXS',
-                            'AKNTS_PWXS',
-                        ],
-                    ],
-                ],
-                'price' => [
-                    [
-                        'locale' => null,
-                        'scope' => 'channel1',
-                        'data' => [['amount' => 10, 'currency' => 'EUR']],
-                    ],
-                    [
-                        'locale' => null,
-                        'scope' => 'channel2',
-                        'data' => [['amount' => 20, 'currency' => 'USD']],
-                    ],
-                ],
-            ],
+            $item,
         ]);
     }
 
-    function it_publishes_association_type_created_event(ProducerInterface $producer)
-    {
-        $this->beConstructedWith($producer, 'message_type');
+    function it_publishes_association_type_created_event(
+        ProducerInterface $producer,
+        EventDispatcherInterface $eventDispatcher
+    ) {
+        $this->beConstructedWith($producer, 'message_type', $eventDispatcher);
 
-        $producer->publish(json_encode([
-            'type' => 'message_type',
-            'payload' => [
+        $item = [
                 'code' => 'X_SELL',
                 'labels' => [
                     'en_US' => 'Cross sell',
                     'fr_FR' => 'Vente croisée',
                 ],
-            ],
+            ];
+
+        $eventDispatcher
+            ->dispatch('akeneo_producer_message.post_publish', new MessageEvent('message_type', $item))
+            ->shouldBeCalled();
+
+        $producer->publish(json_encode([
+            'type' => 'message_type',
+            'payload' => $item,
             'recordedOn' => (new \DateTime())->format('Y-m-d H:i:s'),
         ]))->shouldBeCalled();
 
         $this->write([
-            [
-                'code' => 'X_SELL',
-                'labels' => [
-                    'en_US' => 'Cross sell',
-                    'fr_FR' => 'Vente croisée',
-                ],
-            ],
+            $item,
         ]);
     }
 
-    function it_publishes_category_created_event(ProducerInterface $producer)
+    function it_publishes_category_created_event(ProducerInterface $producer, EventDispatcherInterface $eventDispatcher)
     {
-        $this->beConstructedWith($producer, 'message_type');
+        $this->beConstructedWith($producer, 'message_type', $eventDispatcher);
 
-        $producer->publish(json_encode([
-            'type' => 'message_type',
-            'payload' => [
+        $item = [
                 'code' => 'master',
                 'parent' => null,
                 'labels' => [
                     'en_US' => 'Master catalog',
                     'de_DE' => 'Hauptkatalog',
                     'fr_FR' => 'Catalogue principal'
                 ],
-            ],
+            ];
+
+        $eventDispatcher
+            ->dispatch('akeneo_producer_message.post_publish', new MessageEvent('message_type', $item))
+            ->shouldBeCalled();
+
+        $producer->publish(json_encode([
+            'type' => 'message_type',
+            'payload' => $item,
             'recordedOn' => (new \DateTime())->format('Y-m-d H:i:s'),
         ]))->shouldBeCalled();
 
         $this->write([
-            [
-                'code' => 'master',
-                'parent' => null,
-                'labels' => [
-                    'en_US' => 'Master catalog',
-                    'de_DE' => 'Hauptkatalog',
-                    'fr_FR' => 'Catalogue principal'
-                ],
-            ],
+            $item,
         ]);
     }
 
-    function it_does_not_publish_events_if_there_is_nothing_to_publish(ProducerInterface $producer)
-    {
-        $this->beConstructedWith($producer, 'message_type');
+    function it_does_not_publish_events_if_there_is_nothing_to_publish(
+        ProducerInterface $producer,
+        EventDispatcherInterface $eventDispatcher
+    ) {
+        $this->beConstructedWith($producer, 'message_type', $eventDispatcher);
         $producer->publish(Argument::any())->shouldNotBeCalled();
+        $eventDispatcher->dispatch(Argument::any())->shouldNotBeCalled();
 
         $this->write([]);
     }
+
+    function it_should_dispatch_a_message_for_every_item_published(
+        ProducerInterface $producer,
+        EventDispatcherInterface $eventDispatcher
+    ) {
+        $this->beConstructedWith($producer, 'message_type', $eventDispatcher);
+
+        $item = [
+            'code' => 'master',
+            'parent' => null,
+            'labels' => [
+                'en_US' => 'Master catalog',
+                'de_DE' => 'Hauptkatalog',
+                'fr_FR' => 'Catalogue principal'
+            ],
+        ];
+
+        $items = [
+            $item,
+            $item,
+            $item,
+            $item,
+        ];
+
+        $itemsCount = count($items);
+
+        $producer->publish(json_encode([
+            'type' => 'message_type',
+            'payload' => $item,
+            'recordedOn' => (new \DateTime())->format('Y-m-d H:i:s'),
+        ]))->shouldBeCalledTimes($itemsCount);
+
+        $eventDispatcher
+            ->dispatch('akeneo_producer_message.post_publish', new MessageEvent('message_type', $item))
+            ->shouldBeCalledTimes($itemsCount);
+
+        $this->write($items);
+    }
 }

spec/Event/MessageEventSpec.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace spec\Sylake\AkeneoProducerBundle\Event;
+
+use PhpSpec\ObjectBehavior;
+
+class MessageEventSpec extends ObjectBehavior
+{
+    function it_should_be_constructed_correctly()
+    {
+        $payload = [
+            'foo' => 'bar'
+        ];
+
+        $type = 'type';
+
+        $this->beConstructedWith($type, $payload);
+
+        $this->getType()->shouldReturn($type);
+        $this->getPayload()->shouldReturn($payload);
+    }
+}

src/Connector/Writer/RabbitMqProducer.php

@@ -4,6 +4,8 @@
 
 use Akeneo\Component\Batch\Item\ItemWriterInterface;
 use OldSound\RabbitMqBundle\RabbitMq\ProducerInterface;
+use Sylake\AkeneoProducerBundle\Event\MessageEvent;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
 final class RabbitMqProducer implements ItemWriterInterface
 {
@@ -18,13 +20,20 @@ final class RabbitMqProducer implements ItemWriterInterface
     private $messageType;
 
     /**
-     * @param ProducerInterface $producer
-     * @param string $messageType
+     * @var EventDispatcherInterface
      */
-    public function __construct(ProducerInterface $producer, $messageType)
+    private $eventDispatcher;
+
+    /**
+     * @param ProducerInterface        $producer
+     * @param                          $messageType
+     * @param EventDispatcherInterface $eventDispatcher
+     */
+    public function __construct(ProducerInterface $producer, $messageType, EventDispatcherInterface $eventDispatcher)
     {
         $this->producer = $producer;
         $this->messageType = $messageType;
+        $this->eventDispatcher = $eventDispatcher;
     }
 
     /**
@@ -38,6 +47,11 @@ public function write(array $items)
                 'payload' => $item,
                 'recordedOn' => (new \DateTime())->format('Y-m-d H:i:s'),
             ]));
+
+            $this->eventDispatcher->dispatch(
+                MessageEvent::POST_PUBLISH,
+                new MessageEvent($this->messageType, $item)
+            );
         }
     }
 }

src/Event/MessageEvent.php

@@ -0,0 +1,46 @@
+<?php
+
+namespace Sylake\AkeneoProducerBundle\Event;
+
+use Symfony\Component\EventDispatcher\Event;
+
+class MessageEvent extends Event
+{
+    const POST_PUBLISH = 'akeneo_producer_message.post_publish';
+
+    /**
+     * @var string
+     */
+    protected $type;
+
+    /**
+     * @var mixed
+     */
+    protected $payload;
+
+    /**
+     * @param string $type
+     * @param mixed $payload
+     */
+    public function __construct($type, $payload)
+    {
+        $this->type = $type;
+        $this->payload = $payload;
+    }
+
+    /**
+     * @return string
+     */
+    public function getType()
+    {
+        return $this->type;
+    }
+
+    /**
+     * @return mixed
+     */
+    public function getPayload()
+    {
+        return $this->payload;
+    }
+}