Remove middleware layer (use event-dispatcher). Add eventable tick handler for loop strategy.

Author: ZhukV

CHANGELOG.md

@@ -1,6 +1,12 @@
 CHANGELOG
 =========
 
+Next release
+------------
+
+* Remove consumer middleware layer. Use `symfony/event-dispatcher` instead of.
+* Remove `RamseyUuidMessageIdGenerator` (use `SymfonyUuidMessageIdGenerator` instead). 
+
 v2.2.0
 ------
 

README.md

@@ -51,26 +51,10 @@ Development
 
 For easy development you can use the `Docker`.
 
-> **Note:** We use internal network for link the our library with rabbitmq for testing
-  and development.
-
 ```bash
-docker network create --driver bridge amqp
-docker run -d --rm \
-    --network amqp \
-    --name amqp-rabbitmq \
-    rabbitmq:management
-docker build -t amqp .
-docker run -it --rm \
-    --name amqp \
-    -v $(pwd):/code \
-    --network amqp \
-    -e "RABBITMQ_HOST=amqp-rabbitmq" \
-    amqp bash
-
-```
-
-> **Note** for debugging you can expose 15672 port for access to management plugin. 
+docker compose up
+docker compose exec amqp bash
+``` 
 
 After success run and attach to container you must install vendors:
 

composer.json

@@ -33,7 +33,8 @@
         "guzzlehttp/guzzle": "~7.0",
         "symfony/console": "~6.4 | ~7.0",
         "symfony/process": "~6.4 | ~7.0",
-        "ramsey/uuid": "~4.7",
+        "symfony/uid": "~6.4 | ~7.0",
+        "symfony/event-dispatcher": "~6.4 | ~7.0",
         "fivelab/transactional": "~2.0",
         "fivelab/ci-rules": "dev-master",
         "psr/log": "*",
@@ -44,10 +45,11 @@
 
     "suggest": {
         "ext-amqp": "For use amqp extension.",
-        "ext-sockets": "For php-amqplib",
-        "php-amqplib/php-amqplib": "pure PHP driver",
+        "ext-sockets": "For php-amqplib.",
+        "ext-pcntl": "For catch signals and use alarm.",
+        "php-amqplib/php-amqplib": "pure PHP driver.",
         "symfony/console": "For configure console commands for run consumer.",
-        "ramsey/uuid": "For use UUID message id generator",
+        "symfony/uid": "For use UUID message id generator",
         "fivelab/transactional": "For use transactional layer with this wrapper."
     },
 

docker-compose.yml

@@ -23,4 +23,4 @@ services:
 networks:
     amqp:
         name: fivelab-amqp
-        driver: bridge
+        driver: bridge

docs/basic-usage.md

@@ -102,13 +102,11 @@ And you can create and run consumer:
 <?php
 
 use FiveLab\Component\Amqp\Consumer\SingleConsumer;
-use FiveLab\Component\Amqp\Consumer\Middleware\ConsumerMiddlewares;
 use FiveLab\Component\Amqp\Consumer\ConsumerConfiguration;
 
 $consumer = new SingleConsumer(
     $queueFactory,
     new MyMessageHandler(),
-    new ConsumerMiddlewares(),
     new ConsumerConfiguration()
 );
 

src/Adapter/AmqpLib/Queue/AmqpQueue.php

@@ -42,6 +42,7 @@ public function consume(\Closure $handler, string $tag = ''): void
         $amqplibChannel = $this->channel->getChannel();
 
         $queueName = $this->getName();
+        $stopConsuming = false;
 
         try {
             $amqplibChannel->basic_consume(
@@ -51,15 +52,19 @@ public function consume(\Closure $handler, string $tag = ''): void
                 false,
                 $this->definition->exclusive,
                 false,
-                function (AMQPMessage $message) use ($handler, $queueName) {
+                function (AMQPMessage $message) use ($handler, $queueName, &$stopConsuming): void {
                     $receivedMessage = new AmqpReceivedMessage($message, $queueName);
 
-                    $handler($receivedMessage);
+                    $result = $handler($receivedMessage);
+
+                    if (false === $result) {
+                        $stopConsuming = true;
+                    }
                 }
             );
 
             // Loop as long as the channel has callbacks registered
-            while ($amqplibChannel->is_consuming()) {
+            while (!$stopConsuming && $amqplibChannel->is_consuming()) {
                 $amqplibChannel->wait(null, false, $this->getChannel()->getConnection()->getReadTimeout());
             }
         } catch (\Throwable $e) {

src/Command/OutputEventHandler.php

@@ -17,17 +17,18 @@
 use FiveLab\Component\Amqp\Consumer\Checker\RunConsumerCheckerRegistryInterface;
 use FiveLab\Component\Amqp\Consumer\ConsumerInterface;
 use FiveLab\Component\Amqp\Consumer\EventableConsumerInterface;
-use FiveLab\Component\Amqp\Consumer\Middleware\StopAfterNExecutesMiddleware;
-use FiveLab\Component\Amqp\Consumer\MiddlewareAwareInterface;
 use FiveLab\Component\Amqp\Consumer\Registry\ConsumerRegistryInterface;
+use FiveLab\Component\Amqp\Event\ProcessedMessageEvent;
 use FiveLab\Component\Amqp\Exception\RunConsumerCheckerNotFoundException;
+use FiveLab\Component\Amqp\Listener\StopAfterNExecutesListener;
 use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Command\SignalableCommandInterface;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
 #[AsCommand(name: 'event-broker:consumer:run', description: 'Run consumer.')]
 class RunConsumerCommand extends Command implements SignalableCommandInterface
@@ -37,7 +38,8 @@ class RunConsumerCommand extends Command implements SignalableCommandInterface
 
     public function __construct(
         private readonly ConsumerRegistryInterface $consumerRegistry,
-        ?RunConsumerCheckerRegistryInterface       $runCheckerRegistry = null
+        ?RunConsumerCheckerRegistryInterface       $runCheckerRegistry = null,
+        private readonly ?EventDispatcherInterface $eventDispatcher = null
     ) {
         parent::__construct();
 
@@ -95,21 +97,24 @@ protected function execute(InputInterface $input, OutputInterface $output): int
         $this->consumer = $this->consumerRegistry->get($consumerKey);
 
         if ($this->consumer instanceof EventableConsumerInterface) {
-            $closure = (new OutputEventHandler($output))(...);
-
-            $this->consumer->addEventHandler($closure);
+            $this->consumer->setEventDispatcher($this->eventDispatcher);
         }
 
         if ($input->getOption('messages')) {
-            if (!$this->consumer instanceof MiddlewareAwareInterface) {
+            if (!$this->consumer instanceof EventableConsumerInterface) {
                 throw new \InvalidArgumentException(\sprintf(
                     'For set number of messages customer must implement "%s", but "%s" given.',
-                    MiddlewareAwareInterface::class,
+                    EventableConsumerInterface::class,
                     \get_class($this->consumer)
                 ));
             }
 
-            $this->consumer->pushMiddleware(new StopAfterNExecutesMiddleware((int) $input->getOption('messages')));
+            if (!$this->eventDispatcher) {
+                throw new \RuntimeException('A message limit can\'t be applied, since the command has no access to the event dispatcher.');
+            }
+
+            $listener = new StopAfterNExecutesListener($this->eventDispatcher, (int) $input->getOption('messages'));
+            $this->consumer->getEventDispatcher()?->addListener(ProcessedMessageEvent::class, $listener->onProcessedMessage(...));
         }
 
         if ($input->getOption('read-timeout')) {

src/Command/RunConsumerCommand.php

@@ -13,8 +13,6 @@
 
 namespace FiveLab\Component\Amqp\Command;
 
-use FiveLab\Component\Amqp\Consumer\ConsumerInterface;
-use FiveLab\Component\Amqp\Consumer\Event;
 use FiveLab\Component\Amqp\Consumer\RoundRobin\RoundRobinConsumer;
 use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
@@ -41,10 +39,7 @@ public function getSubscribedSignals(): array
             return [];
         }
 
-        return [
-            \SIGINT,
-            \SIGTERM,
-        ];
+        return [\SIGINT, \SIGTERM];
     }
 
     public function handleSignal(int $signal, false|int $previousExitCode = 0): int|false
@@ -54,21 +49,6 @@ public function handleSignal(int $signal, false|int $previousExitCode = 0): int|
         return false;
     }
 
-    protected function initialize(InputInterface $input, OutputInterface $output): void
-    {
-        $this->consumer->addEventHandler(static function (Event $event, mixed ...$args) use ($output) {
-            if (Event::ChangeConsumer === $event) {
-                /** @var ConsumerInterface $consumer */
-                $consumer = $args[0];
-
-                $output->writeln(\sprintf(
-                    'Select next consumer with queue <comment>%s</comment>.',
-                    $consumer->getQueue()->getName()
-                ), OutputInterface::VERBOSITY_VERBOSE);
-            }
-        });
-    }
-
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
         $this->consumer->run();

src/Command/RunRoundRobinConsumerCommand.php

@@ -78,7 +78,6 @@ public function connect(): void
             }
         }
 
-        // @phpstan-ignore-next-line
         throw $firstException;
     }