Add consume strategy. Add handle signals.

Author: ZhukV

CHANGELOG.md

@@ -1,6 +1,13 @@
 CHANGELOG
 =========
 
+Next release
+------------
+
+* Set minimum Symfony version to 6.4
+* Remove `--loop` option. Use `LoopConsumer` instead.
+* Add consume strategy (`Default` and `Loop`).
+
 v2.1.3
 ------
 
@@ -15,7 +22,7 @@ v2.1.1
 ------
 
 * Add queue name to `ReceivedMessage`.
-* Add method `ReceivedMessage::isDirectPublisher` for check, is message direct published to queue (via default exchange).
+* Add method `ReceivedMessage::isDirectPublished` for check, is message direct published to queue (via default exchange).
 
 v2.1.0
 ------

Dockerfile

@@ -1,6 +1,6 @@
 FROM php:8.4-cli
 
-MAINTAINER Vitalii Zhuk <v.zhuk@fivelab.org>
+LABEL org.opencontainers.image.authors="Vitalii Zhuk <v.zhuk@fivelab.org>"
 
 ARG XDEBUG_REMOTE_HOST='host.docker.internal'
 ARG XDEBUG_REMOTE_PORT=9000
@@ -19,7 +19,7 @@ RUN \
         librabbitmq-dev && \
     printf '\n' | pecl install amqp && \
     yes | pecl install xdebug && \
-    docker-php-ext-install sockets && \
+    docker-php-ext-install sockets pcntl && \
     docker-php-ext-enable amqp xdebug
 
 # Configure XDebug

composer.json

@@ -25,12 +25,14 @@
     "require-dev": {
         "ext-amqp": "*",
         "ext-sockets": "*",
+        "ext-pcntl": "*",
         "phpunit/phpunit": "~11.5",
         "phpmetrics/phpmetrics": "~3.0",
         "phpstan/phpstan": "~2.1.6",
         "escapestudios/symfony2-coding-standard": "~3.5",
         "guzzlehttp/guzzle": "~7.0",
-        "symfony/console": "~5.4 | ~6.0 | ~7.0",
+        "symfony/console": "~6.4 | ~7.0",
+        "symfony/process": "~6.4 | ~7.0",
         "ramsey/uuid": "~4.7",
         "fivelab/transactional": "~2.0",
         "fivelab/ci-rules": "dev-master",

docker-compose.yml

@@ -0,0 +1,26 @@
+services:
+    rabbitmq:
+        image: rabbitmq:management
+        container_name: fivelab-amqp-rabbitmq
+        ports:
+            - '15672:15672'
+        networks:
+            - amqp
+
+    amqp:
+        build:
+            dockerfile: Dockerfile
+            context: .
+        container_name: fivelab-amqp-php
+        tty: true
+        volumes:
+            - './:/code'
+        environment:
+            RABBITMQ_HOST: fivelab-amqp-rabbitmq
+        networks:
+            - amqp
+
+networks:
+    amqp:
+        name: fivelab-amqp
+        driver: bridge

phpstan.neon

@@ -2,8 +2,3 @@ parameters:
     level: 8
     paths:
         - src
-
-    ignoreErrors:
-        -
-            message: '#^Property FiveLab\\Component\\Amqp\\Command\\.+::\$(defaultName|defaultDescription) has no type specified.$#'
-            path: src/

src/Adapter/Amqp/Queue/AmqpQueue.php

@@ -44,7 +44,7 @@ public function consume(\Closure $handler, string $tag = ''): void
             $this->queue->consume(function (\AMQPEnvelope $envelope) use ($handler) {
                 $receivedMessage = new AmqpReceivedMessage($this->queue, $envelope);
 
-                $handler($receivedMessage);
+                return $handler($receivedMessage);
             }, AMQP_NOPARAM, $tag);
         } catch (\AMQPQueueException $e) {
             if (false !== \stripos($e->getMessage(), 'consumer timeout exceed')) {

src/AmqpBuilder.php

@@ -0,0 +1,83 @@
+<?php
+
+/*
+ * This file is part of the FiveLab Amqp 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\Component\Amqp;
+
+use FiveLab\Component\Amqp\Adapter\Amqp\Channel\AmqpChannelFactory;
+use FiveLab\Component\Amqp\Adapter\Amqp\Connection\AmqpConnectionFactory;
+use FiveLab\Component\Amqp\Adapter\Amqp\Exchange\AmqpExchangeFactory;
+use FiveLab\Component\Amqp\Adapter\Amqp\Queue\AmqpQueueFactory;
+use FiveLab\Component\Amqp\Channel\ChannelFactoryInterface;
+use FiveLab\Component\Amqp\Channel\Definition\ChannelDefinition;
+use FiveLab\Component\Amqp\Connection\ConnectionFactoryInterface;
+use FiveLab\Component\Amqp\Connection\Dsn;
+use FiveLab\Component\Amqp\Exchange\Definition\ExchangeDefinition;
+use FiveLab\Component\Amqp\Exchange\ExchangeFactoryInterface;
+use FiveLab\Component\Amqp\Publisher\Middleware\PublisherMiddlewares;
+use FiveLab\Component\Amqp\Publisher\Publisher;
+use FiveLab\Component\Amqp\Publisher\PublisherInterface;
+use FiveLab\Component\Amqp\Queue\Definition\QueueDefinition;
+use FiveLab\Component\Amqp\Queue\QueueFactoryInterface;
+
+class AmqpBuilder
+{
+    private readonly Dsn $dsn;
+    private ?ConnectionFactoryInterface $connectionFactory = null;
+    private ?ChannelFactoryInterface $channelFactory = null;
+
+    public function __construct(Dsn|string $dsn)
+    {
+        if (\is_string($dsn)) {
+            $dsn = Dsn::fromDsn($dsn);
+        }
+
+        $this->dsn = $dsn;
+    }
+
+    public function createConnection(): ConnectionFactoryInterface
+    {
+        if (!$this->connectionFactory) {
+            $this->connectionFactory = new AmqpConnectionFactory($this->dsn);
+        }
+
+        return $this->connectionFactory;
+    }
+
+    public function createChannel(): ChannelFactoryInterface
+    {
+        if (!$this->channelFactory) {
+            $this->channelFactory = new AmqpChannelFactory($this->createConnection(), new ChannelDefinition());
+        }
+
+        return $this->channelFactory;
+    }
+
+    public function createExchange(ExchangeDefinition $definition): ExchangeFactoryInterface
+    {
+        return new AmqpExchangeFactory($this->createChannel(), $definition);
+    }
+
+    public function createPublisher(ExchangeDefinition $definition, ?PublisherMiddlewares $middlewares = null): PublisherInterface
+    {
+        $exchange = $this->createExchange($definition);
+
+        return new Publisher($exchange, $middlewares ?: new PublisherMiddlewares());
+    }
+
+    public function createQueue(QueueDefinition $definition): QueueFactoryInterface
+    {
+        $channel = $this->createChannel();
+
+        return new AmqpQueueFactory($channel, $definition);
+    }
+}

src/Command/InitializeExchangesCommand.php

@@ -22,8 +22,6 @@
 #[AsCommand(name: 'event-broker:initialize:exchanges', description: 'Initialize exchanges.')]
 class InitializeExchangesCommand extends Command
 {
-    protected static $defaultName = 'event-broker:initialize:exchanges';
-    protected static $defaultDescription = 'Initialize exchanges.';
     private ExchangeFactoryRegistryInterface $registry;
 
     /**
@@ -45,12 +43,6 @@ public function __construct(ExchangeFactoryRegistryInterface $registry, array $e
         $this->exchanges = $exchanges;
     }
 
-    protected function configure(): void
-    {
-        $this
-            ->setDescription(self::$defaultDescription);
-    }
-
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
         foreach ($this->exchanges as $exchange) {

src/Command/InitializeQueuesCommand.php

@@ -22,8 +22,6 @@
 #[AsCommand(name: 'event-broker:initialize:queues', description: 'Initialize queues.')]
 class InitializeQueuesCommand extends Command
 {
-    protected static $defaultName = 'event-broker:initialize:queues';
-    protected static $defaultDescription = 'Initialize queues.';
     private QueueFactoryRegistryInterface $registry;
 
     /**
@@ -45,12 +43,6 @@ public function __construct(QueueFactoryRegistryInterface $registry, array $queu
         $this->queues = $queues;
     }
 
-    protected function configure(): void
-    {
-        $this
-            ->setDescription(self::$defaultDescription);
-    }
-
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
         foreach ($this->queues as $queue) {

src/Command/ListConsumersCommand.php

@@ -21,9 +21,6 @@
 #[AsCommand(name: 'event-broker:consumer:list', description: 'List of possible consumers.')]
 class ListConsumersCommand extends Command
 {
-    protected static $defaultName = 'event-broker:consumer:list';
-    protected static $defaultDescription = 'List of possible consumers.';
-
     /**
      * @var array<string>
      */
@@ -41,12 +38,6 @@ public function __construct(array $consumers)
         $this->consumers = $consumers;
     }
 
-    protected function configure(): void
-    {
-        $this
-            ->setDescription(self::$defaultDescription);
-    }
-
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
         $output->writeln('Possible consumers are:');