Added option to change strategy for publishing events

Author: Nyholm

composer.json

@@ -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

@@ -100,5 +100,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']);
+    }
+}