Merge pull request #13 from driftphp/feature/command-bus-distribution

mmoreram · web-flow · commit c52dc6861e2a · 2020-02-03T18:09:49.000+01:00
Added alternative command_bus distribution
diff --git a/Bus/Bus.php b/Bus/Bus.php
@@ -17,32 +17,57 @@
 
 use Drift\CommandBus\Exception\InvalidCommandException;
 use Drift\CommandBus\Middleware\DebugableMiddleware;
-use Drift\CommandBus\Middleware\Middleware;
+use React\EventLoop\LoopInterface;
+use React\Promise\Deferred;
 
 /**
  * Interface Bus.
  */
 abstract class Bus
 {
+    /**
+     * @var string
+     */
+    const DISTRIBUTION_INLINE = 'inline';
+
+    /**
+     * @var string
+     */
+    const DISTRIBUTION_NEXT_TICK = 'next_tick';
+
     /**
      * @var callable
      */
     private $middlewareChain;
 
+    /**
+     * @var LoopInterface
+     */
+    private $loop;
+
     /**
      * @var array
      */
     private $middleware;
 
     /**
-     * @param array $middleware
+     * @param LoopInterface $loop
+     * @param array         $middleware
+     * @param string        $distribution
      */
-    public function __construct(array $middleware)
-    {
+    public function __construct(
+        LoopInterface $loop,
+        array $middleware,
+        string $distribution
+    ) {
+        $this->loop = $loop;
         $this->middleware = array_map(function (DebugableMiddleware $middleware) {
             return $middleware->getMiddlewareInfo();
         }, $middleware);
-        $this->middlewareChain = $this->createExecutionChain($middleware);
+
+        $this->middlewareChain = self::DISTRIBUTION_NEXT_TICK === $distribution
+            ? $this->createNextTickExecutionChain($middleware)
+            : $this->createInlineExecutionChain($middleware);
     }
 
     /**
@@ -64,13 +89,13 @@ protected function handle($command)
     }
 
     /**
-     * Create execution chain.
+     * Create inline execution chain.
      *
      * @param array $middlewareList
      *
      * @return callable
      */
-    private function createExecutionChain($middlewareList)
+    private function createInlineExecutionChain($middlewareList)
     {
         $lastCallable = function () {};
 
@@ -83,6 +108,36 @@ private function createExecutionChain($middlewareList)
         return $lastCallable;
     }
 
+    /**
+     * Create next tick execution chain.
+     *
+     * @param array $middlewareList
+     *
+     * @return callable
+     */
+    private function createNextTickExecutionChain($middlewareList)
+    {
+        $lastCallable = function () {};
+
+        while ($middleware = array_pop($middlewareList)) {
+            $lastCallable = function ($command) use ($middleware, $lastCallable) {
+                $deferred = new Deferred();
+                $this
+                    ->loop
+                    ->futureTick(function () use ($deferred, $middleware, $command, $lastCallable) {
+                        $deferred->resolve($middleware->execute(
+                            $command,
+                            $lastCallable
+                        ));
+                    });
+
+                return $deferred->promise();
+            };
+        }
+
+        return $lastCallable;
+    }
+
     /**
      * Get middleware list.
      *
diff --git a/DependencyInjection/CommandBusConfiguration.php b/DependencyInjection/CommandBusConfiguration.php
@@ -15,6 +15,7 @@
 
 namespace Drift\CommandBus\DependencyInjection;
 
+use Drift\CommandBus\Bus\Bus;
 use Mmoreram\BaseBundle\DependencyInjection\BaseConfiguration;
 use Symfony\Component\Config\Definition\Builder\ArrayNodeDefinition;
 
@@ -36,6 +37,10 @@ protected function setupTree(ArrayNodeDefinition $rootNode)
                 ->arrayNode('query_bus')
                     ->addDefaultsIfNotSet()
                     ->children()
+                        ->enumNode('distribution')
+                            ->values([Bus::DISTRIBUTION_INLINE, Bus::DISTRIBUTION_NEXT_TICK])
+                            ->defaultValue(Bus::DISTRIBUTION_INLINE)
+                        ->end()
                         ->arrayNode('middlewares')
                             ->scalarPrototype()
                                 ->defaultValue([])
@@ -47,6 +52,10 @@ protected function setupTree(ArrayNodeDefinition $rootNode)
                 ->arrayNode('command_bus')
                     ->addDefaultsIfNotSet()
                     ->children()
+                        ->enumNode('distribution')
+                            ->values([Bus::DISTRIBUTION_INLINE, Bus::DISTRIBUTION_NEXT_TICK])
+                            ->defaultValue(Bus::DISTRIBUTION_INLINE)
+                        ->end()
                         ->arrayNode('middlewares')
                             ->scalarPrototype()
                                 ->defaultValue([])
diff --git a/DependencyInjection/CommandBusExtension.php b/DependencyInjection/CommandBusExtension.php
@@ -65,7 +65,9 @@ protected function getParametrizationValues(array $config): array
     {
         return [
             'bus.query_bus.middlewares' => $config['query_bus']['middlewares'],
+            'bus.query_bus.distribution' => $config['query_bus']['distribution'],
             'bus.command_bus.middlewares' => $config['command_bus']['middlewares'],
+            'bus.command_bus.distribution' => $config['command_bus']['distribution'],
             'bus.command_bus.async_adapter' => $config['command_bus']['async_adapter'] ?? false,
         ];
     }
diff --git a/DependencyInjection/CompilerPass/BusCompilerPass.php b/DependencyInjection/CompilerPass/BusCompilerPass.php
@@ -28,6 +28,7 @@
 use Drift\CommandBus\Middleware\AsyncMiddleware;
 use Drift\CommandBus\Middleware\HandlerMiddleware;
 use Drift\CommandBus\Middleware\Middleware;
+use React\EventLoop\LoopInterface;
 use ReflectionClass;
 use ReflectionException;
 use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
@@ -172,11 +173,13 @@ private function createQueryBus(ContainerBuilder $container)
     {
         $container->setDefinition('drift.query_bus', (new Definition(
             QueryBus::class, [
+                new Reference(LoopInterface::class),
                 $this->createMiddlewaresArray(
                     $container,
                     'query',
                     false
                 ),
+                $container->getParameter('bus.query_bus.distribution'),
             ]
         ))
             ->addTag('preload')
@@ -198,11 +201,13 @@ private function createCommandBus(
     ) {
         $container->setDefinition('drift.command_bus', (new Definition(
             CommandBus::class, [
+                new Reference(LoopInterface::class),
                 $this->createMiddlewaresArray(
                     $container,
                     'command',
                     $asyncBus
                 ),
+                $container->getParameter('bus.command_bus.distribution'),
             ]
         ))
             ->addTag('preload')
@@ -221,11 +226,13 @@ private function createInlineCommandBus(ContainerBuilder $container)
     {
         $container->setDefinition('drift.inline_command_bus', (new Definition(
             InlineCommandBus::class, [
+                new Reference(LoopInterface::class),
                 $this->createMiddlewaresArray(
                     $container,
                     'command',
                     false
                 ),
+                $container->getParameter('bus.command_bus.distribution'),
             ]
         ))
             ->addTag('preload')
diff --git a/Tests/Bus/CommandHandlerTest.php b/Tests/Bus/CommandHandlerTest.php
@@ -15,6 +15,7 @@
 
 namespace Drift\CommandBus\Tests\Bus;
 
+use Drift\CommandBus\Bus\Bus;
 use Drift\CommandBus\Middleware\HandlerMiddleware;
 use Drift\CommandBus\Tests\BusFunctionalTest;
 use Drift\CommandBus\Tests\Command\ChangeAThing;
@@ -44,9 +45,23 @@ protected static function decorateConfiguration(array $configuration): array
             ],
         ];
 
+        $configuration['command_bus']['command_bus']['distribution'] = self::distributedBus()
+            ? Bus::DISTRIBUTION_NEXT_TICK
+            : Bus::DISTRIBUTION_INLINE;
+
         return $configuration;
     }
 
+    /**
+     * Create distributed bus.
+     *
+     * @return bool
+     */
+    protected static function distributedBus(): bool
+    {
+        return false;
+    }
+
     /**
      * Test buses are being built.
      */
diff --git a/Tests/Bus/DistributedCommandHandlerTest.php b/Tests/Bus/DistributedCommandHandlerTest.php
@@ -0,0 +1,32 @@
+<?php
+
+/*
+ * This file is part of the DriftPHP Project
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ *
+ * Feel free to edit as you please, and have fun.
+ *
+ * @author Marc Morera <yuhu@mmoreram.com>
+ */
+
+declare(strict_types=1);
+
+namespace Drift\CommandBus\Tests\Bus;
+
+/**
+ * Class DistributedCommandHandlerTest.
+ */
+class DistributedCommandHandlerTest extends CommandHandlerTest
+{
+    /**
+     * Create distributed bus.
+     *
+     * @return bool
+     */
+    protected static function distributedBus(): bool
+    {
+        return true;
+    }
+}
diff --git a/Tests/Bus/DistributedQueryHandlerTest.php b/Tests/Bus/DistributedQueryHandlerTest.php
@@ -0,0 +1,32 @@
+<?php
+
+/*
+ * This file is part of the DriftPHP Project
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ *
+ * Feel free to edit as you please, and have fun.
+ *
+ * @author Marc Morera <yuhu@mmoreram.com>
+ */
+
+declare(strict_types=1);
+
+namespace Drift\CommandBus\Tests\Bus;
+
+/**
+ * Class DistributedQueryHandlerTest.
+ */
+class DistributedQueryHandlerTest extends QueryHandlerTest
+{
+    /**
+     * Create distributed bus.
+     *
+     * @return bool
+     */
+    protected static function distributedBus(): bool
+    {
+        return true;
+    }
+}
diff --git a/Tests/Bus/QueryHandlerTest.php b/Tests/Bus/QueryHandlerTest.php
@@ -15,6 +15,7 @@
 
 namespace Drift\CommandBus\Tests\Bus;
 
+use Drift\CommandBus\Bus\Bus;
 use Drift\CommandBus\Exception\InvalidCommandException;
 use Drift\CommandBus\Tests\BusFunctionalTest;
 use Drift\CommandBus\Tests\Context;
@@ -44,9 +45,23 @@ protected static function decorateConfiguration(array $configuration): array
             ],
         ];
 
+        $configuration['command_bus']['query_bus']['distribution'] = self::distributedBus()
+            ? Bus::DISTRIBUTION_NEXT_TICK
+            : Bus::DISTRIBUTION_INLINE;
+
         return $configuration;
     }
 
+    /**
+     * Create distributed bus.
+     *
+     * @return bool
+     */
+    protected static function distributedBus(): bool
+    {
+        return false;
+    }
+
     /**
      * Test buses are being built.
      */

Original file line number	Diff line number	Diff line change
`@@ -65,7 +65,9 @@ protected function getParametrizationValues(array $config): array`
`65`	`65`	`{`
`66`	`66`	`return [`
`67`	`67`	`'bus.query_bus.middlewares' => $config['query_bus']['middlewares'],`
	`68`	`+ 'bus.query_bus.distribution' => $config['query_bus']['distribution'],`
`68`	`69`	`'bus.command_bus.middlewares' => $config['command_bus']['middlewares'],`
	`70`	`+ 'bus.command_bus.distribution' => $config['command_bus']['distribution'],`
`69`	`71`	`'bus.command_bus.async_adapter' => $config['command_bus']['async_adapter'] ?? false,`
`70`	`72`	`];`
`71`	`73`	`}`