Bump service bus package

.travis.yml

@@ -14,10 +14,11 @@ cache:
 matrix:
   fast_finish: true
   include:
-    - php: 7
+    - php: 7.1
       env:
+        - DEPENDENCIES=""
         - EXECUTE_CS_CHECK=true
-        - EXECUTE_TEST_COVERALLS=true
+        - TEST_COVERAGE=true
 
 before_install:
   - if [[ $EXECUTE_TEST_COVERALLS != 'true' ]]; then phpenv config-rm xdebug.ini || return 0 ; fi
@@ -29,14 +30,11 @@ install:
   - composer info -i
 
 script:
-  - if [[ $EXECUTE_TEST_COVERALLS == 'true' ]]; then php ./vendor/bin/phpunit --coverage-text --coverage-clover ./build/logs/clover.xml ; fi
-  - if [[ $EXECUTE_TEST_COVERALLS != 'true' ]]; then ./vendor/bin/phpunit ; fi
-  - if [[ $EXECUTE_CS_CHECK == 'true' ]]; then ./vendor/bin/php-cs-fixer fix -v --diff --dry-run ; fi
+  - if [[ $TEST_COVERAGE == 'true' ]]; then php -dzend_extension=xdebug.so ./vendor/bin/phpunit --coverage-text --coverage-clover ./build/logs/clover.xml; else ./vendor/bin/phpunit; fi
+  - if [[ $EXECUTE_CS_CHECK == 'true' ]]; then ./vendor/bin/php-cs-fixer fix -v --diff --dry-run; fi
 
 after_script:
-  - if [[ $EXECUTE_TEST_COVERALLS == 'true' ]]; then ./vendor/bin/coveralls ; fi
-  - wget https://scrutinizer-ci.com/ocular.phar
-  - if [[ $EXECUTE_TEST_COVERALLS == 'true' ]]; then php ocular.phar code-coverage:upload --format=php-clover build/logs/clover.xml ; fi
+  - if [[ $TEST_COVERAGE == 'true' ]]; then php vendor/bin/coveralls -v; fi
 
 notifications:
   webhooks:

composer.json

@@ -50,28 +50,29 @@
     }
   },
   "require": {
-    "php": "^7.0",
+    "php": "^7.1",
     "symfony/config": "~2.8 || ~3.0",
     "symfony/dependency-injection": "~2.8 || ~3.0",
     "symfony/http-kernel": "~2.8 || ~3.0",
-    "prooph/service-bus": "^5.1",
+    "prooph/service-bus": "~6.0",
     "container-interop/container-interop": "^1.1"
   },
   "require-dev": {
     "phpunit/phpunit": "^5.3",
     "phpunit/php-invoker": "^1.1.4",
     "fabpot/php-cs-fixer": "^1.11",
     "bookdown/bookdown": "1.x-dev as 1.0.0",
-    "tobiju/bookdown-bootswatch-templates": "1.0.x-dev"
+    "tobiju/bookdown-bootswatch-templates": "^1.0",
+    "satooshi/php-coveralls": "^1.0",
+    "phpspec/prophecy": "dev-patch-1 as 1.6.2"
   },
   "suggest": {
   },
   "minimum-stability": "dev",
   "prefer-stable": true,
   "extra": {
     "branch-alias": {
-      "dev-master": "1.0-dev",
-      "dev-develop": "1.1-dev"
+      "dev-master": "1.0-dev"
     }
   },
   "scripts": {
@@ -94,5 +95,11 @@
       "phpunit.xml*",
       "test"
     ]
-  }
+  },
+  "repositories": [
+    {
+      "type": "git",
+      "url": "https://github.com/prolic/prophecy.git"
+    }
+  ]
 }

src/DependencyInjection/Compiler/PluginsPass.php

@@ -12,7 +12,6 @@
 namespace Prooph\Bundle\ServiceBus\DependencyInjection\Compiler;
 
 use Prooph\Bundle\ServiceBus\DependencyInjection\ProophServiceBusExtension;
-use Symfony\Component\DependencyInjection\Reference;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
 
@@ -32,12 +31,13 @@ public function process(ContainerBuilder $container)
                 $typePlugins = $container->findTaggedServiceIds(sprintf('prooph_service_bus.%s.plugin', $type . '_bus'));
                 $plugins = $container->findTaggedServiceIds(sprintf('prooph_service_bus.%s.plugin', $name));
 
-                $plugins = array_merge($globalPlugins, $typePlugins, $plugins);
+                $plugins = array_merge(array_keys($globalPlugins), array_keys($typePlugins), array_keys($plugins));
 
-                foreach ($plugins as $id => $args) {
-                    $definition = $container->findDefinition('prooph_service_bus.' . $name);
-                    $definition->addMethodCall('utilize', [new Reference($id)]);
-                }
+                $busDefinition = $container->getDefinition($bus);
+                $busPlugins = $busDefinition->getArgument(2);
+
+                $finalPlugins = array_merge($busPlugins, $plugins);
+                $busDefinition->replaceArgument(2, $finalPlugins);
             }
         }
     }

src/DependencyInjection/ProophServiceBusExtension.php

@@ -11,12 +11,13 @@
 
 namespace Prooph\Bundle\ServiceBus\DependencyInjection;
 
+use Prooph\Bundle\ServiceBus\Exception\RuntimeException;
 use Prooph\ServiceBus\CommandBus;
 use Prooph\ServiceBus\EventBus;
-use Prooph\ServiceBus\QueryBus;
 use Prooph\ServiceBus\Plugin\Router\CommandRouter;
 use Prooph\ServiceBus\Plugin\Router\EventRouter;
 use Prooph\ServiceBus\Plugin\Router\QueryRouter;
+use Prooph\ServiceBus\QueryBus;
 use Symfony\Component\Config\FileLocator;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\DependencyInjection\DefinitionDecorator;
@@ -35,6 +36,7 @@ final class ProophServiceBusExtension extends Extension
         'query' => QueryBus::class,
     ];
 
+
     public function getNamespace()
     {
         return 'http://getprooph.org/schemas/symfony-dic/prooph';
@@ -118,51 +120,64 @@ private function busLoad(
      */
     private function loadBus(string $type, string $name, array $options, ContainerBuilder $container)
     {
+        $serviceBusName = sprintf('prooph_service_bus.%s', $name);
         $serviceBusDefinition = $container->setDefinition(
-            sprintf('prooph_service_bus.%s', $name),
+            $serviceBusName,
             new DefinitionDecorator('prooph_service_bus.' . $type . '_bus')
         );
 
         if (!empty($options['plugins'])) {
             foreach ($options['plugins'] as $index => $util) {
-                $serviceBusDefinition->addMethodCall('utilize', [new Reference($util)]);
+                if (! $container->hasDefinition($util)) {
+                    throw new RuntimeException(
+                        "A plugin must be a string representing an configured container service"
+                    );
+                }
             }
         }
-        // define message factory
-        $messageFactoryId = 'prooph_service_bus.message_factory.' . $name;
 
-        $container
-            ->setDefinition(
+        // define message factory
+        $messageFactoryId = 'prooph_service_bus.message_factory.'.$name;
+        $container->setDefinition(
                 $messageFactoryId,
                 new DefinitionDecorator($options['message_factory'])
             );
 
+
         // define message factory plugin
-        $messageFactoryPluginId = 'prooph_service_bus.message_factory_plugin.' . $name;
+        $messageFactoryPluginId = 'prooph_service_bus.message_factory_plugin.'.$name;
+        $messageFactoryPluginDefinition = new DefinitionDecorator('prooph_service_bus.message_factory_plugin');
+        $messageFactoryPluginDefinition->setArguments([new Reference($messageFactoryId)]);
 
-        $container
-            ->setDefinition(
+        $container->setDefinition(
                 $messageFactoryPluginId,
-                new DefinitionDecorator('prooph_service_bus.message_factory_plugin')
-            )
-            ->setArguments([new Reference($messageFactoryId)]);
+                $messageFactoryPluginDefinition
+            );
 
-        $serviceBusDefinition->addMethodCall('utilize', [new Reference($messageFactoryPluginId)]);
 
         // define router
+        $routerId = null;
         if (!empty($options['router'])) {
             $routerId = sprintf('prooph_service_bus.%s.router', $name);
-
-            $routerDefinition = $container->setDefinition(
-                $routerId,
-                new DefinitionDecorator($options['router']['type'])
-            );
+            $routerDefinition = new DefinitionDecorator($options['router']['type']);
             $routerDefinition->setArguments([$options['router']['routes'] ?? []]);
-
-            $serviceBusDefinition->addMethodCall('utilize', [new Reference($routerId)]);
+            $container->setDefinition($routerId, $routerDefinition);
         }
 
-        //Add container plugin
-        $serviceBusDefinition->addMethodCall('utilize', [new Reference('prooph_service_bus.container_plugin')]);
+        //Attach container plugin
+        $containerPluginId = 'prooph_service_bus.container_plugin';
+        $pluginIds = array_filter(array_merge($options['plugins'], [$containerPluginId, $messageFactoryPluginId, $routerId]));
+
+
+        // Wrap the message bus creation into factory to call attachToMessageBus on the plugins
+        $serviceBusDefinition
+            ->setFactory([new Reference('prooph_service_bus.'.$type.'_bus_factory'), 'create'])
+            ->setArguments(
+                [
+                    $container->getDefinition('prooph_service_bus.'.$type.'_bus')->getClass(),
+                    new Reference('service_container'),
+                    $pluginIds,
+                ]
+            );
     }
 }

src/MessageBusFactory.php

@@ -0,0 +1,31 @@
+<?php
+/**
+ * prooph (http://getprooph.org/)
+ *
+ * @see       https://github.com/prooph/service-bus-symfony-bundle for the canonical source repository
+ * @copyright Copyright (c) 2017 prooph software GmbH (http://prooph-software.com/)
+ * @license   https://github.com/prooph/service-bus-symfony-bundle/blob/master/LICENSE.md New BSD License
+ */
+
+declare(strict_types=1);
+
+namespace Prooph\Bundle\ServiceBus;
+
+use Prooph\ServiceBus\MessageBus;
+use Symfony\Component\DependencyInjection\ContainerInterface;
+
+class MessageBusFactory
+{
+    public function create(string $class, ContainerInterface $container, array $plugins = []) : MessageBus
+    {
+        /** @var MessageBus $bus */
+        $bus = new $class();
+
+        foreach ($plugins as $pluginId) {
+            $plugin = $container->get($pluginId);
+            $plugin->attachToMessageBus($bus);
+        }
+
+        return $bus;
+    }
+}

src/Resources/config/command_bus.xml

@@ -10,7 +10,10 @@
     </parameters>
 
     <services>
-        <service id="prooph_service_bus.command_bus" class="%prooph_service_bus.command_bus.class%" abstract="true" />
+        <service id="prooph_service_bus.command_bus" class="%prooph_service_bus.command_bus.class%" abstract="true" >
+            <factory service="prooph_service_bus.command_bus_factory" method="create"/>
+        </service>
+        <service id="prooph_service_bus.command_bus_factory" class="%prooph_service_bus.message_bus_factory.class%"/>
         <service id="prooph_service_bus.command_bus_router" class="%prooph_service_bus.command_bus_router.class%" public="false" abstract="true" />
     </services>
 </container>

src/Resources/config/event_bus.xml

@@ -10,7 +10,10 @@
     </parameters>
 
     <services>
-        <service id="prooph_service_bus.event_bus" class="%prooph_service_bus.event_bus.class%" abstract="true" />
+        <service id="prooph_service_bus.event_bus" class="%prooph_service_bus.event_bus.class%" abstract="true" >
+            <factory service="prooph_service_bus.event_bus_factory" method="create"/>
+        </service>
+        <service id="prooph_service_bus.event_bus_factory" class="%prooph_service_bus.message_bus_factory.class%"/>
         <service id="prooph_service_bus.event_bus_router" class="%prooph_service_bus.event_bus_router.class%" public="false" abstract="true" />
     </services>
 </container>

src/Resources/config/query_bus.xml

@@ -10,7 +10,10 @@
     </parameters>
 
     <services>
-        <service id="prooph_service_bus.query_bus" class="%prooph_service_bus.query_bus.class%" abstract="true" />
+        <service id="prooph_service_bus.query_bus" class="%prooph_service_bus.query_bus.class%" abstract="true" >
+            <factory service="prooph_service_bus.query_bus_factory" method="create"/>
+        </service>
+        <service id="prooph_service_bus.query_bus_factory" class="%prooph_service_bus.message_bus_factory.class%"/>
         <service id="prooph_service_bus.query_bus_router" class="%prooph_service_bus.query_bus_router.class%" public="false" abstract="true" />
     </services>
 </container>

src/Resources/config/service_bus.xml

@@ -5,6 +5,7 @@
     xsi:schemaLocation="http://symfony.com/schema/dic/services http://symfony.com/schema/dic/services/services-1.0.xsd">
 
     <parameters>
+        <parameter key="prooph_service_bus.message_bus_factory.class">Prooph\Bundle\ServiceBus\MessageBusFactory</parameter>
         <parameter key="prooph_service_bus.message_factory_plugin.class">Prooph\ServiceBus\Plugin\MessageFactoryPlugin</parameter>
         <parameter key="prooph_service_bus.message_factory.class">Prooph\Common\Messaging\FQCNMessageFactory</parameter>
         <parameter key="prooph_service_bus.container_plugin.class">Prooph\ServiceBus\Plugin\ServiceLocatorPlugin</parameter>

test/DependencyInjection/Fixture/Model/AcmeRegisterUserCommand.php

@@ -19,7 +19,7 @@ final class AcmeRegisterUserCommand extends Command implements PayloadConstructa
 {
     use PayloadTrait;
 
-    public function messageName()
+    public function messageName() : string
     {
         return 'Acme\RegisterUser';
     }

test/DependencyInjection/Fixture/Model/AcmeRegisterUserHandler.php

@@ -20,6 +20,11 @@ public function __invoke(AcmeRegisterUserCommand $command)
         $this->lastCommand = $command;
     }
 
+    public function handle(AcmeRegisterUserCommand $command)
+    {
+        $this->lastCommand = $command;
+    }
+
     public function lastCommand()
     {
         return $this->lastCommand;

test/DependencyInjection/Fixture/Model/AcmeUserEventListener.php

@@ -15,7 +15,12 @@ class AcmeUserEventListener
 {
     private $lastEvent;
 
-    public function onUserWasRegistered(AcmeUserWasRegisteredEvent $event)
+    public function __invoke(AcmeUserWasRegisteredEvent $event)
+    {
+        $this->lastEvent = $event;
+    }
+
+    public function onEvent(AcmeUserWasRegisteredEvent $event)
     {
         $this->lastEvent = $event;
     }

test/DependencyInjection/Fixture/Model/AcmeUserWasRegisteredEvent.php

@@ -19,7 +19,7 @@ class AcmeUserWasRegisteredEvent extends DomainEvent implements PayloadConstruct
 {
     use PayloadTrait;
 
-    public function messageName()
+    public function messageName() : string
     {
         return 'Acme\UserWasRegistered';
     }

test/DependencyInjection/Fixture/Model/MockPlugin.php

@@ -12,11 +12,11 @@
 
 use Prooph\Common\Event\ActionEvent;
 use Prooph\Common\Event\ActionEventEmitter;
-use Prooph\Common\Event\ActionEventListenerAggregate;
 use Prooph\Common\Event\DetachAggregateHandlers;
 use Prooph\ServiceBus\MessageBus;
+use Prooph\ServiceBus\Plugin\AbstractPlugin;
 
-class MockPlugin implements ActionEventListenerAggregate
+class MockPlugin extends AbstractPlugin
 {
     use DetachAggregateHandlers;
 
@@ -30,9 +30,13 @@ public function wasFired(): bool
     /**
      * @param ActionEventEmitter $dispatcher
      */
-    public function attach(ActionEventEmitter $dispatcher)
+    public function attachToMessageBus(MessageBus $bus) : void
     {
-        $this->trackHandler($dispatcher->attachListener(MessageBus::EVENT_INITIALIZE, [$this, 'onInitialize']));
+        $this->trackHandler($bus->attach(
+            MessageBus::EVENT_DISPATCH,
+            [$this, 'onInitialize'],
+            MessageBus::PRIORITY_INITIALIZE
+        ));
     }
 
     public function onInitialize(ActionEvent $event)