Update amqp package. Add listeners to configuration. Remove middlewares/event_handlers configurations.

Author: ZhukV

.github/workflows/tests.yaml

@@ -57,6 +57,9 @@ jobs:
                     -   php: 8.4
                         symfony: '~7.0'
 
+                    -   php: 8.5
+                        symfony: '~7.0'
+
 
         steps:
             -   name: Checkout

Dockerfile

@@ -12,6 +12,7 @@ RUN \
 RUN \
     apt-get install -y --no-install-recommends \
     yes | pecl install xdebug && \
+    docker-php-ext-install pcntl && \
     docker-php-ext-enable xdebug
 
 # Install composer

composer.json

@@ -20,18 +20,20 @@
 
     "require": {
         "php": "^8.2",
-        "fivelab/amqp": "~2.2.0",
-        "symfony/dependency-injection": "~6.4 | ~7.0",
-        "symfony/framework-bundle": "~6.4 | ~7.0"
+        "fivelab/amqp": "dev-master",
+        "symfony/dependency-injection": "~6.4 | ~7.0 | ~8.0",
+        "symfony/framework-bundle": "~6.4 | ~7.0 | ~8.0"
     },
 
     "require-dev": {
         "phpunit/phpunit": "~11.5",
         "phpmetrics/phpmetrics": "^3.0@rc",
         "escapestudios/symfony2-coding-standard": "~3.5",
         "matthiasnoback/symfony-dependency-injection-test": "~6.0",
-        "symfony/console": "~6.4 | ~7.0",
-        "symfony/expression-language": "~6.4 | ~7.0",
+        "symfony/console": "~6.4 | ~7.0 | 8.0",
+        "symfony/expression-language": "~6.4 | ~7.0 | 8.0",
+        "doctrine/persistence": "~4.0",
+        "doctrine/dbal": "~4.0",
         "fivelab/ci-rules": "dev-master"
     },
 

src/DependencyInjection/AmqpExtension.php

@@ -14,6 +14,8 @@
 namespace FiveLab\Bundle\AmqpBundle\DependencyInjection;
 
 use FiveLab\Bundle\AmqpBundle\Factory\DriverFactory;
+use FiveLab\Bundle\AmqpBundle\Listener\PingDbalConnectionsListener;
+use FiveLab\Bundle\AmqpBundle\Listener\ReleaseMemoryListener;
 use FiveLab\Component\Amqp\Channel\ChannelFactoryInterface;
 use FiveLab\Component\Amqp\Connection\ConnectionFactoryInterface;
 use FiveLab\Component\Amqp\Connection\Dsn;
@@ -33,7 +35,6 @@
 use FiveLab\Component\Amqp\Queue\Definition\Arguments\SingleActiveCustomerArgument;
 use FiveLab\Component\Amqp\Queue\QueueFactoryInterface;
 use Symfony\Component\Config\FileLocator;
-use Symfony\Component\DependencyInjection\Argument\ServiceClosureArgument;
 use Symfony\Component\DependencyInjection\ChildDefinition;
 use Symfony\Component\DependencyInjection\Compiler\ServiceLocatorTagPass;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
@@ -71,14 +72,16 @@ public function load(array $configs, ContainerBuilder $container): void
         $loader->load('definitions.php');
         $loader->load('consumers.php');
         $loader->load('publishers.php');
+        $loader->load('listeners.php');
         $loader->load('services.php');
 
+        $this->configureListeners($container, $config['listeners']);
         $this->configureConnections($container, $config['connections']);
         $this->configureChannels($container, $config['channels']);
         $this->configureExchanges($container, $config['exchanges']);
         $this->configureQueues($container, $config['queues'], $config['queue_default_arguments'] ?? []);
         $this->configurePublishers($container, $config['publishers'], $config['publisher_middleware']);
-        $this->configureConsumers($container, $config['consumers'], $config['consumer_middleware'], $config['consumer_event_handlers'], $config['consumer_defaults']);
+        $this->configureConsumers($container, $config['consumers'], $config['consumer_defaults']);
 
         if ($this->isConfigEnabled($container, $config['round_robin'])) {
             $loader->load('round-robin.php');
@@ -93,8 +96,6 @@ public function load(array $configs, ContainerBuilder $container): void
                 $container,
                 $config['delay'],
                 $config['publisher_middleware'],
-                $config['consumer_middleware'],
-                $config['consumer_event_handlers'],
                 $config['queue_default_arguments'] ?? [],
                 $config['consumer_defaults'] ?? []
             );
@@ -132,6 +133,21 @@ private function setDefaultStrategy(array $config): array
         return $config;
     }
 
+    private function configureListeners(ContainerBuilder $container, array $listeners): void
+    {
+        if (null !== $listeners['release_memory']) {
+            $container->getDefinition(ReleaseMemoryListener::class)
+                ->setAbstract(false)
+                ->replaceArgument(1, $listeners['release_memory']);
+        }
+
+        if (null !== $listeners['ping_dbal_connections']) {
+            $container->getDefinition(PingDbalConnectionsListener::class)
+                ->setAbstract(false)
+                ->replaceArgument(1, $listeners['ping_dbal_connections']);
+        }
+    }
+
     private function configureConnections(ContainerBuilder $container, array $connections): void
     {
         $registryDef = $container->getDefinition('fivelab.amqp.connection_factory_registry');
@@ -517,7 +533,7 @@ private function configureQueues(ContainerBuilder $container, array $queues, arr
         $container->setParameter('fivelab.amqp.queue_factories', \array_keys($this->queueFactories));
     }
 
-    private function configureConsumers(ContainerBuilder $container, array $consumers, array $globalMiddlewares, array $eventHandlers, array $defaults): void
+    private function configureConsumers(ContainerBuilder $container, array $consumers, array $defaults): void
     {
         $consumerRegistryDef = $container->getDefinition('fivelab.amqp.consumer_registry');
         $checkConsumerRegistryDef = $container->getDefinition('fivelab.amqp.consumer_checker_registry');
@@ -574,19 +590,6 @@ private function configureConsumers(ContainerBuilder $container, array $consumer
                 $queueFactoryServiceId = $this->queueFactories[$consumer['queue']];
             }
 
-            // Configure middleware for consumer
-            $consumerMiddlewares = \array_merge($globalMiddlewares, $consumer['middleware']);
-
-            $middlewareList = \array_map(static function (string $serviceId) {
-                return new Reference($serviceId);
-            }, $consumerMiddlewares);
-
-            $middlewareServiceId = \sprintf('fivelab.amqp.consumer.%s.middlewares', $key);
-            $middlewareServiceDef = new ChildDefinition('fivelab.amqp.consumer.middlewares.abstract');
-            $middlewareServiceDef->setArguments($middlewareList);
-
-            $container->setDefinition($middlewareServiceId, $middlewareServiceDef);
-
             // Configure message handler for consumer
             $messageHandlersServiceId = \sprintf('fivelab.amqp.consumer.%s.message_handler', $key);
             $messageHandlersServiceDef = new ChildDefinition('fivelab.amqp.consumer.message_handler.abstract');
@@ -642,9 +645,8 @@ private function configureConsumers(ContainerBuilder $container, array $consumer
                 $consumerServiceDef
                     ->replaceArgument(0, new Reference($queueFactoryServiceId))
                     ->replaceArgument(1, new Reference($messageHandlersServiceId))
-                    ->replaceArgument(2, new Reference($middlewareServiceId))
-                    ->replaceArgument(3, new Reference($consumerConfigurationServiceId))
-                    ->replaceArgument(4, new Reference($strategyServiceId));
+                    ->replaceArgument(2, new Reference($consumerConfigurationServiceId))
+                    ->replaceArgument(3, new Reference($strategyServiceId));
             } elseif ('spool' === $consumer['mode']) {
                 // Configure spool consumer
                 $consumerConfigurationServiceId = \sprintf('fivelab.amqp.consumer.%s.configuration', $key);
@@ -662,9 +664,8 @@ private function configureConsumers(ContainerBuilder $container, array $consumer
                 $consumerServiceDef
                     ->replaceArgument(0, new Reference($queueFactoryServiceId))
                     ->replaceArgument(1, new Reference($messageHandlersServiceId))
-                    ->replaceArgument(2, new Reference($middlewareServiceId))
-                    ->replaceArgument(3, new Reference($consumerConfigurationServiceId))
-                    ->replaceArgument(4, new Reference($strategyServiceId));
+                    ->replaceArgument(2, new Reference($consumerConfigurationServiceId))
+                    ->replaceArgument(3, new Reference($strategyServiceId));
             } elseif ('loop' === $consumer['mode']) {
                 $consumerConfigurationServiceId = \sprintf('fivelab.amqp.consumer.%s.configuration', $key);
                 $consumerConfigurationServiceDef = new ChildDefinition('fivelab.amqp.consumer_loop.configuration.abstract');
@@ -680,23 +681,15 @@ private function configureConsumers(ContainerBuilder $container, array $consumer
                 $consumerServiceDef
                     ->replaceArgument(0, new Reference($queueFactoryServiceId))
                     ->replaceArgument(1, new Reference($messageHandlersServiceId))
-                    ->replaceArgument(2, new Reference($middlewareServiceId))
-                    ->replaceArgument(3, new Reference($consumerConfigurationServiceId))
-                    ->replaceArgument(4, new Reference($strategyServiceId));
+                    ->replaceArgument(2, new Reference($consumerConfigurationServiceId))
+                    ->replaceArgument(3, new Reference($strategyServiceId));
             } else {
                 throw new \InvalidArgumentException(\sprintf(
                     'Unknown mode "%s".',
                     $consumer['mode']
                 ));
             }
 
-            foreach ($eventHandlers as $eventHandler) {
-                $consumerServiceDef->addMethodCall(
-                    'addEventHandler',
-                    [new ServiceClosureArgument(new Reference($eventHandler)), true]
-                );
-            }
-
             $container->setDefinition($consumerConfigurationServiceId, $consumerConfigurationServiceDef);
             $container->setDefinition($consumerServiceId, $consumerServiceDef);
 
@@ -830,7 +823,7 @@ private function configureRoundRobin(ContainerBuilder $container, array $config)
             ->replaceArgument(2, \array_keys($this->consumers));
     }
 
-    private function configureDelay(ContainerBuilder $container, array $config, array $globalPublisherMiddlewares, array $globalConsumerMiddlewares, array $consumerEventHandlers, array $queueDefaultArguments, array $consumerDefaults): void
+    private function configureDelay(ContainerBuilder $container, array $config, array $globalPublisherMiddlewares, array $queueDefaultArguments, array $consumerDefaults): void
     {
         // Configure exchange
         $this->configureExchanges($container, [
@@ -949,7 +942,7 @@ private function configureDelay(ContainerBuilder $container, array $config, arra
                     'idle_timeout'     => 100000,
                 ],
             ],
-        ], $globalConsumerMiddlewares, $consumerEventHandlers, $consumerDefaults);
+        ], $consumerDefaults);
     }
 
     private function createArgumentDefinition(ContainerBuilder $container, string $serviceId, string $class, mixed ...$values): Reference

src/DependencyInjection/Configuration.php

@@ -28,6 +28,7 @@ public function getConfigTreeBuilder(): TreeBuilder
 
         $rootNode
             ->children()
+                ->append($this->getListenersDefinition())
                 ->append($this->getDelayDefinition())
                 ->append($this->getRoundRobinDefinition())
                 ->append($this->getConnectionsNodeDefinition())
@@ -37,15 +38,42 @@ public function getConfigTreeBuilder(): TreeBuilder
                 ->append($this->getQueueArgumentsNodeDefinition('queue_default_arguments'))
                 ->append($this->getConsumersNodeDefinition())
                 ->append($this->getConsumerDefaults())
-                ->append($this->getConsumerEventHandlersNodeDefinition())
                 ->append($this->getPublishersNodeDefinition())
-                ->append($this->getMiddlewareNodeDefinition('consumer_'))
                 ->append($this->getMiddlewareNodeDefinition('publisher_'))
             ->end();
 
         return $treeBuilder;
     }
 
+    private function getListenersDefinition(): ArrayNodeDefinition
+    {
+        $node = new ArrayNodeDefinition('listeners');
+
+        $node
+            ->addDefaultsIfNotSet()
+            ->children()
+                ->scalarNode('release_memory')
+                    ->defaultValue(null)
+                    ->info('Enable release memory listener (true - clear before handle, false - after handle, null - disable listener).')
+                    ->validate()
+                        ->ifFalse(static fn (mixed $value) => null === $value || \is_bool($value))
+                        ->thenInvalid('Invalid value for "release_memory". Must be bool or null.')
+                    ->end()
+                ->end()
+
+                ->scalarNode('ping_dbal_connections')
+                    ->defaultValue(null)
+                    ->info('Enable ping DBAL connections listeners (number - seconds for ping interval, null - disable listener)')
+                    ->validate()
+                        ->ifFalse(static fn (mixed $value) => null === $value || \is_int($value) || $value < 1)
+                        ->thenInvalid('Invalid value for "ping_dbal_connections". Must be integer or null.')
+                    ->end()
+                ->end()
+            ->end();
+
+        return $node;
+    }
+
     private function getDelayDefinition(): NodeDefinition
     {
         $node = new ArrayNodeDefinition('delay');
@@ -407,8 +435,6 @@ private function getConsumersNodeDefinition(): NodeDefinition
                     ->end()
                 ->end()
 
-                ->append($this->getMiddlewareNodeDefinition())
-
                 ->arrayNode('options')
                     ->addDefaultsIfNotSet()
                     ->children()
@@ -662,18 +688,6 @@ private function getMiddlewareNodeDefinition(string $nodePrefix = ''): NodeDefin
         return $node;
     }
 
-    private function getConsumerEventHandlersNodeDefinition(): NodeDefinition
-    {
-        $node = new ArrayNodeDefinition('consumer_event_handlers');
-
-        $node
-            ->defaultValue([])
-            ->prototype('scalar')
-            ->end();
-
-        return $node;
-    }
-
     private function getStrategyNodeDefinition(?string $defaultValue = null): NodeDefinition
     {
         return (new ScalarNodeDefinition('strategy'))

src/Listener/PingDbalConnectionsListener.php

@@ -0,0 +1,80 @@
+<?php
+
+/*
+ * This file is part of the FiveLab AmqpBundle package
+ *
+ * (c) FiveLab
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code
+ */
+
+declare(strict_types = 1);
+
+namespace FiveLab\Bundle\AmqpBundle\Listener;
+
+use Doctrine\DBAL\Connection;
+use Doctrine\Persistence\ConnectionRegistry;
+use FiveLab\Component\Amqp\Event\ConsumerTickEvent;
+use Symfony\Component\Console\ConsoleEvents;
+use Symfony\Component\Console\Event\ConsoleSignalEvent;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+
+readonly class PingDbalConnectionsListener implements EventSubscriberInterface
+{
+    private \ArrayObject $options;
+
+    public function __construct(private ConnectionRegistry $registry, private int $interval, private array $signals = [\SIGALRM])
+    {
+        $this->options = new \ArrayObject([
+            'ping'      => false,
+            'last_ping' => \time(),
+            'output'    => null,
+        ]);
+    }
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            ConsoleEvents::SIGNAL    => ['onConsoleSignal', 0],
+            ConsumerTickEvent::class => ['onConsumerTick', 0],
+        ];
+    }
+
+    public function onConsoleSignal(ConsoleSignalEvent $event): void
+    {
+        if (!\in_array($event->getHandlingSignal(), $this->signals, true)) {
+            return;
+        }
+
+        $nextPing = $this->options['last_ping'] + $this->interval;
+
+        if ($nextPing < \time()) {
+            $this->options->offsetSet('ping', true);
+            $this->options->offsetSet('output', $event->getOutput());
+        }
+    }
+
+    public function onConsumerTick(): void
+    {
+        if (!$this->options['ping']) {
+            return;
+        }
+
+        $this->options->offsetSet('ping', false);
+        $this->options->offsetSet('last_ping', \time());
+
+        /** @var Connection $connection */
+        foreach ($this->registry->getConnections() as $key => $connection) {
+            if ($connection->isConnected()) {
+                $connection->executeQuery('SELECT 1');
+
+                $this->options['output']->writeln(\sprintf(
+                    'Ping <comment>%s</comment> database connection.',
+                    $key
+                ), OutputInterface::VERBOSITY_DEBUG);
+            }
+        }
+    }
+}