Merge pull request #6 from Nyholm/predefined

Added option to change strategy for publishing events

Author: matthiasnoback

composer.json

@@ -16,7 +16,7 @@
         "php": ">=5.4",
         "beberlei/assert": "~2.0",
         "simple-bus/message-bus": "~2.0",
-        "simple-bus/asynchronous": "~2.0",
+        "simple-bus/asynchronous": "~2.1",
         "simple-bus/symfony-bridge": "~4.0",
         "symfony/framework-bundle": "~2.0|~3.0"
     },
@@ -26,7 +26,8 @@
     "require-dev": {
         "phpunit/phpunit": "~4.0",
         "symfony/monolog-bundle": "~2.3",
-        "satooshi/php-coveralls": "~0.6"
+        "satooshi/php-coveralls": "~0.6",
+        "matthiasnoback/symfony-dependency-injection-test": "^0.7"
     },
     "autoload": {
         "psr-4" : { "SimpleBus\\AsynchronousBundle\\" : "src" }

doc/getting_started.md

@@ -68,3 +68,15 @@ simple_bus_asynchronous:
 ```
 
 This will log consumed messages to the `asynchronous_command_bus` and `asynchronous_event_bus` channels respectively.
+
+## Choose event strategy
+
+When handling events you have two strategies to choose from. Either you publish *all* events to the message queue or
+you only publish the events that have a registered subscriber. If your application is the only one that consuming messages
+you should consider using the **predefined** strategy. This will reduce the message overhead on the message queue. 
+
+```yaml
+simple_bus_asynchronous:
+  events:
+    strategy: 'predefined' # default: 'always'
+```

src/DependencyInjection/Compiler/CollectAsynchronousEventNames.php

@@ -0,0 +1,37 @@
+<?php
+
+namespace SimpleBus\AsynchronousBundle\DependencyInjection\Compiler;
+
+use SimpleBus\SymfonyBridge\DependencyInjection\Compiler\CollectServices;
+use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+
+class CollectAsynchronousEventNames implements CompilerPassInterface
+{
+    use CollectServices;
+
+    /**
+     * Get all asynchronous event subscribers and save the name of the event they are listening to.
+     *
+     * @param ContainerBuilder $container
+     */
+    public function process(ContainerBuilder $container)
+    {
+        $serviceId = 'simple_bus.asynchronous.publishes_predefined_messages_middleware';
+        if (!$container->hasDefinition($serviceId)) {
+            return;
+        }
+
+        $names = array();
+        $this->collectServiceIds(
+            $container,
+            'asynchronous_event_subscriber',
+            'subscribes_to',
+            function ($key) use (&$names) {
+                $names[] = $key;
+            }
+        );
+
+        $container->getDefinition($serviceId)->replaceArgument(2, array_unique($names));
+    }
+}

src/DependencyInjection/Configuration.php

@@ -43,6 +43,11 @@ public function getConfigTreeBuilder()
                 ->arrayNode('events')
                     ->canBeEnabled()
                     ->children()
+                        ->enumNode('strategy')
+                            ->info('What strategy to use to publish messages')
+                            ->defaultValue('always')
+                            ->values(['always', 'predefined'])
+                        ->end()
                         ->scalarNode('publisher_service_id')
                             ->info('Service id of an instance of Publisher')
                             ->isRequired()

src/DependencyInjection/SimpleBusAsynchronousExtension.php

@@ -53,12 +53,7 @@ private function requireBundle($bundleName, ContainerBuilder $container)
     {
         $enabledBundles = $container->getParameter('kernel.bundles');
         if (!isset($enabledBundles[$bundleName])) {
-            throw new \LogicException(
-                sprintf(
-                    'You need to enable "%s" as well',
-                    $bundleName
-                )
-            );
+            throw new \LogicException(sprintf('You need to enable "%s" as well', $bundleName));
         }
     }
 
@@ -100,5 +95,15 @@ private function loadAsynchronousEventBus(array $config, ContainerBuilder $conta
         if ($config['logging']['enabled']) {
             $loader->load('asynchronous_events_logging.yml');
         }
+
+
+        if ($config['strategy'] === 'always') {
+            $eventMiddleware = 'simple_bus.asynchronous.always_publishes_messages_middleware';
+        } else {
+            $eventMiddleware = 'simple_bus.asynchronous.publishes_predefined_messages_middleware';
+        }
+
+        // insert before the middleware that actually notifies a message subscriber of the message
+        $container->getDefinition($eventMiddleware)->addTag('event_bus_middleware', ['priority' => 0]);
     }
 }

src/Resources/config/asynchronous_events.yml

@@ -13,9 +13,14 @@ services:
         public: false
         arguments:
             - '@simple_bus.asynchronous.event_publisher'
-        tags:
-            # insert before the middleware that actually notifies a message subscriber of the message
-            - { name: event_bus_middleware, priority: 0 }
+
+    simple_bus.asynchronous.publishes_predefined_messages_middleware:
+        class: SimpleBus\Asynchronous\MessageBus\PublishesPredefinedMessages
+        public: false
+        arguments:
+            - '@simple_bus.asynchronous.event_publisher'
+            - '@simple_bus.event_bus.event_name_resolver'
+            - []
 
     simple_bus.asynchronous.event_bus.finishes_message_before_handling_next_middleware:
         class: SimpleBus\Message\Bus\Middleware\FinishesHandlingMessageBeforeHandlingNext

src/SimpleBusAsynchronousBundle.php

@@ -2,6 +2,7 @@
 
 namespace SimpleBus\AsynchronousBundle;
 
+use SimpleBus\AsynchronousBundle\DependencyInjection\Compiler\CollectAsynchronousEventNames;
 use SimpleBus\AsynchronousBundle\DependencyInjection\SimpleBusAsynchronousExtension;
 use SimpleBus\SymfonyBridge\DependencyInjection\Compiler\ConfigureMiddlewares;
 use SimpleBus\SymfonyBridge\DependencyInjection\Compiler\RegisterHandlers;
@@ -39,5 +40,7 @@ public function build(ContainerBuilder $container)
                 'subscribes_to'
             )
         );
+
+        $container->addCompilerPass(new CollectAsynchronousEventNames());
     }
 }

tests/Unit/DepencencyInjection/Compiler/CollectAsynchronousEventNamesTest.php

@@ -0,0 +1,38 @@
+<?php
+
+namespace SimpleBus\AsynchronousBundle\Tests\Unit\DependencyInjection\Compiler;
+
+
+use Matthias\SymfonyDependencyInjectionTest\PhpUnit\AbstractCompilerPassTestCase;
+use SimpleBus\AsynchronousBundle\DependencyInjection\Compiler\CollectAsynchronousEventNames;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\DependencyInjection\Definition;
+
+class CollectAsynchronousEventNamesTest extends AbstractCompilerPassTestCase
+{
+    protected function registerCompilerPass(ContainerBuilder $container)
+    {
+        $container->addCompilerPass(new CollectAsynchronousEventNames());
+    }
+
+    /**
+     * @test
+     */
+    public function if_compiler_pass_collects_event_names()
+    {
+        $serviceId = 'simple_bus.asynchronous.publishes_predefined_messages_middleware';
+        $middleware = new Definition();
+        $middleware->addArgument('arg0');
+        $middleware->addArgument('arg1');
+        $middleware->addArgument([]);
+        $this->setDefinition($serviceId, $middleware);
+
+        $subscriber = new Definition();
+        $subscriber->addTag('asynchronous_event_subscriber', ['subscribes_to'=>'foo']);
+        $this->setDefinition('event_subscriber', $subscriber);
+
+        $this->compile();
+
+        $this->assertContainerBuilderHasServiceDefinitionWithArgument($serviceId, 2, ['foo']);
+    }
+}

tests/Unit/DepencencyInjection/SimpleBusAsynchronousExtensionTest.php

@@ -0,0 +1,68 @@
+<?php
+
+namespace SimpleBus\AsynchronousBundle\Tests\Unit\DependencyInjection;
+
+
+use Matthias\SymfonyDependencyInjectionTest\PhpUnit\AbstractExtensionTestCase;
+use SimpleBus\AsynchronousBundle\DependencyInjection\SimpleBusAsynchronousExtension;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+
+class SimpleBusAsynchronousExtensionTest extends AbstractExtensionTestCase
+{
+    protected function getContainerExtensions()
+    {
+        return array(
+            new SimpleBusAsynchronousExtension('simple_bus_asynchronous')
+        );
+    }
+
+    protected function getMinimalConfiguration()
+    {
+        return ['object_serializer_service_id'=>'my_serializer', 'commands'=>['publisher_service_id'=>'pusher'], 'events'=>['publisher_service_id'=>'pusher']];
+    }
+
+
+    /**
+     * @test
+     */
+    public function it_uses_strategy_allways_by_default()
+    {
+        $this->container->setParameter('kernel.bundles', ['SimpleBusCommandBusBundle'=>true, 'SimpleBusEventBusBundle'=>true]);
+        $this->load();
+
+        $this->assertContainerBuilderHasServiceDefinitionWithTag('simple_bus.asynchronous.always_publishes_messages_middleware', 'event_bus_middleware', ['priority'=>0]);
+    }
+
+    /**
+     * @test
+     */
+    public function it_uses_strategy_predefined_when_configured()
+    {
+        $this->container->setParameter('kernel.bundles', ['SimpleBusCommandBusBundle'=>true, 'SimpleBusEventBusBundle'=>true]);
+        $this->load(['events'=>['strategy'=>'predefined']]);
+
+        $this->assertContainerBuilderHasServiceDefinitionWithTag('simple_bus.asynchronous.publishes_predefined_messages_middleware', 'event_bus_middleware', ['priority'=>0]);
+    }
+    
+    /**
+     * @test
+     * @expectedException \LogicException
+     * @expectedExceptionMessageRegExp ".*SimpleBusCommandBusBundle.*"
+     */
+    public function it_throws_exception_if_command_bus_bundle_is_missing()
+    {
+        $this->container->setParameter('kernel.bundles', ['SimpleBusEventBusBundle'=>true]);
+        $this->load(['events'=>['strategy'=>'predefined']]);
+    }
+    
+    /**
+     * @test
+     * @expectedException \LogicException
+     * @expectedExceptionMessageRegExp ".*SimpleBusEventBusBundle.*"
+     */
+    public function it_throws_exception_if_event_bus_bundle_is_missing()
+    {
+        $this->container->setParameter('kernel.bundles', ['SimpleBusCommandBusBundle'=>true]);
+        $this->load(['events'=>['strategy'=>'predefined']]);
+    }
+}