Merge pull request #12 from mtrudu/master

Add RabbitMq Command bus

Author: shouze

DependencyInjection/Configuration.php

@@ -51,6 +51,12 @@ public function getConfigTreeBuilder()
                                     ->append($this->createConsumersNodeDefinition())
                                 ->end()
                             ->end()
+                            ->arrayNode('rabbitmq')
+                                ->children()
+                                    ->scalarNode('producer_guesser')->defaultValue('rezzza_command_bus.old_sound_rabbit.producer_guesser')->end()
+                                    ->scalarNode('consumer_bus')->defaultNull()->end()
+                                ->end()
+                            ->end()
                             ->arrayNode('service')
                                 ->children()
                                     ->scalarNode('id')->isRequired()->end()

DependencyInjection/RezzzaCommandBusExtension.php

@@ -53,6 +53,10 @@ private function createBus($name, array $config, ContainerBuilder $container, $l
                 $this->createSncRedisBusCommandBus($commandBusServiceName, $config, $container);
                 $this->decorateBus($commandBusServiceName, $container, $loggerNormalizer);
                 break;
+            case 'rabbitmq':
+                $this->createRabbitMqBusCommandBus($commandBusServiceName, $config, $container);
+                $this->decorateBus($commandBusServiceName, $container, $loggerNormalizer);
+                break;
             case 'service':
                 $container->setAlias($commandBusServiceName, $config['id']);
                 break;
@@ -62,6 +66,25 @@ private function createBus($name, array $config, ContainerBuilder $container, $l
         }
     }
 
+    private function createRabbitMqBusCommandBus($commandBusServiceName, $config, ContainerBuilder $container)
+    {
+        $service = new Definition('%rezzza_command_bus.old_sound_rabbit_bus.class%', [
+            new Reference($config['producer_guesser']),
+        ]);
+
+        $container->setDefinition($commandBusServiceName, $service);
+
+        if ($config['consumer_bus'] !== null) {
+            $consumerDefinition = new Definition('%rezzza_command_bus.old_sound_rabbit_bus.consumer.class%',
+                [
+                    new Reference($this->getCommandBusServiceName($config['consumer_bus'])),
+                ]
+            );
+
+            $container->setDefinition(sprintf('rezzza_command_bus.old_sound_rabbit_bus.consumer.%s', $config['consumer_bus']), $consumerDefinition);
+        }
+    }
+
     private function createSncRedisBusCommandBus($commandBusServiceName, $config, ContainerBuilder $container)
     {
         $client       = new Reference(sprintf('snc_redis.%s_client', $config['client']));

Provider/OldSoundRabbit/DefaultConsumer.php

@@ -0,0 +1,33 @@
+<?php
+
+namespace Rezzza\CommandBusBundle\Provider\OldSoundRabbit;
+
+use OldSound\RabbitMqBundle\RabbitMq\ConsumerInterface;
+use PhpAmqpLib\Message\AMQPMessage;
+use Rezzza\CommandBus\Domain\CommandBusInterface;
+
+/**
+ * Class DefaultConsumer
+ */
+class DefaultConsumer implements ConsumerInterface
+{
+    protected $bus;
+
+    /**
+     * Constructor.
+     *
+     * @param CommandBusInterface $bus
+     */
+    public function __construct(CommandBusInterface $bus)
+    {
+        $this->bus = $bus;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function execute(AMQPMessage $command)
+    {
+        $this->bus->handle(unserialize($command->body));
+    }
+}

Provider/OldSoundRabbit/NoProducerFoundException.php

@@ -0,0 +1,18 @@
+<?php
+
+namespace Rezzza\CommandBusBundle\Provider\OldSoundRabbit;
+
+use Rezzza\CommandBus\Domain\CommandInterface;
+
+/**
+ * Class NoProducerFoundException
+ */
+class NoProducerFoundException extends \LogicException
+{
+    public function __construct(CommandInterface $command)
+    {
+        $message = sprintf('Producer not found for Command [%s]', get_class($command));
+
+        parent::__construct($message);
+    }
+}

README.md

@@ -51,7 +51,7 @@ $bus->handle(new FooCommand());
 Bus
 ---
 
-Direct & SncRedis are currently supported.
+Direct, SncRedis & OldSoundRabbit are currently supported.
 
 Commands
 --------
@@ -126,6 +126,14 @@ rezzza_command_bus:
     buses:
       synchronous: direct
       asynchronous:
+        old_rabbitmq:
+          #define producer_guesser which allow to determine rigth producer for each command
+          #producer name and command class name must be indentical 
+            #example:
+            #producer name : source_entry_update
+            #command class name : SourceEntryUpdateCommand
+          producer_guesser: rezzza_command_bus.old_sound_rabbit.producer_guesser 
+          consumer_bus: synchronous #consumer handle command with synchronous bus
         snc_redis:
             client: default # snc redis client.
             read_block_timeout: 1 # see blpop documentation

Resources/config/services/services.xml

@@ -7,6 +7,11 @@
     <parameters>
         <parameter key="rezzza_command_bus.direct_bus.class">Rezzza\CommandBus\Infra\Provider\Direct\DirectBus</parameter>
         <parameter key="rezzza_command_bus.snc_redis_bus.class">Rezzza\CommandBusBundle\Provider\SncRedis\SncRedisBus</parameter>
+
+        <parameter key="rezzza_command_bus.old_sound_rabbit_bus.class">Rezzza\CommandBus\Infra\Provider\OldSoundRabbit\RabbitMqBus</parameter>
+        <parameter key="rezzza_command_bus.old_sound_rabbit_bus.consumer.class">Rezzza\CommandBusBundle\Provider\OldSoundRabbit\DefaultConsumer</parameter>
+        <parameter key="rezzza_command_bus.old_sound_rabbit.producer_guesser.class">Rezzza\CommandBus\Infra\Provider\OldSoundRabbit\ProducerGuesser</parameter>
+
         <parameter key="rezzza_command_bus.event_dispatcher_bus.class">Rezzza\CommandBus\Domain\EventDispatcherBus</parameter>
         <parameter key="rezzza_command_bus.logger_bus.class">Rezzza\CommandBus\Domain\LoggerBus</parameter>
 
@@ -32,6 +37,10 @@
     </parameters>
 
     <services>
+        <service id="rezzza_command_bus.old_sound_rabbit.producer_guesser" class="%rezzza_command_bus.old_sound_rabbit.producer_guesser.class%" public="false">
+            <argument type="service" id="service_container" />
+        </service>
+
         <service id="rezzza_command_bus.command_handler_locator.container" class="%rezzza_command_bus.command_handler_locator.container.class%" public="false">
             <argument type="service" id="service_container" />
         </service>