feat: fix types and refactoring

Author: thomasvargiu

composer.json

@@ -49,7 +49,8 @@
     },
     "extra": {
         "laminas": {
-            "module": "RabbitMqModule"
+            "component": "RabbitMqModule",
+            "config-provider": "RabbitMqModule\\ConfigProvider"
         }
     },
     "scripts": {

config/module.config.php

@@ -1,6 +1,6 @@
 <?xml version="1.0"?>
 <psalm
-    errorLevel="5"
+    errorLevel="1"
     resolveFromConfigFile="true"
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xmlns="https://getpsalm.org/schema/config"

psalm.xml

@@ -14,30 +14,20 @@
 
 abstract class BaseAmqp implements SetupFabricAwareInterface
 {
-    /** @var AbstractConnection */
-    protected $connection;
+    protected AbstractConnection $connection;
 
-    /** @var AMQPChannel|null */
-    private $channel;
+    private ?AMQPChannel $channel = null;
 
-    /** @var QueueOptions */
-    protected $queueOptions;
+    protected ?QueueOptions $queueOptions = null;
 
-    /** @var ExchangeOptions */
-    protected $exchangeOptions;
+    protected ?ExchangeOptions $exchangeOptions = null;
 
-    /** @var bool */
-    protected $autoSetupFabricEnabled = true;
+    protected bool $autoSetupFabricEnabled = true;
 
-    /** @var bool */
-    protected $exchangeDeclared = false;
+    protected bool $exchangeDeclared = false;
 
-    /** @var bool */
-    protected $queueDeclared = false;
+    protected bool $queueDeclared = false;
 
-    /**
-     * @param AMQPChannel        $channel
-     */
     public function __construct(AbstractConnection $connection, AMQPChannel $channel = null)
     {
         $this->connection = $connection;
@@ -68,17 +58,17 @@ public function getQueueOptions(): ?QueueOptions
         return $this->queueOptions;
     }
 
-    public function setQueueOptions(QueueOptions $queueOptions): void
+    public function setQueueOptions(?QueueOptions $queueOptions): void
     {
         $this->queueOptions = $queueOptions;
     }
 
-    public function getExchangeOptions(): ExchangeOptions
+    public function getExchangeOptions(): ?ExchangeOptions
     {
         return $this->exchangeOptions;
     }
 
-    public function setExchangeOptions(ExchangeOptions $exchangeOptions): void
+    public function setExchangeOptions(?ExchangeOptions $exchangeOptions): void
     {
         $this->exchangeOptions = $exchangeOptions;
     }
@@ -95,16 +85,14 @@ public function setAutoSetupFabricEnabled(bool $autoSetupFabricEnabled): void
 
     /**
      * Declare Exchange
-     *
-     * @param ExchangeOptions $options
      */
-    protected function declareExchange(ExchangeOptions $options = null): void
+    protected function declareExchange(?ExchangeOptions $options = null): void
     {
         if (! $options) {
             $options = $this->getExchangeOptions();
         }
 
-        if (! $options->isDeclare()) {
+        if (! $options || ! $options->isDeclare()) {
             return;
         }
 
@@ -117,7 +105,7 @@ protected function declareExchange(ExchangeOptions $options = null): void
             $options->isDurable(),
             $options->isAutoDelete(),
             $options->isInternal(),
-            $options->isNoWait(),
+            false,
             $arguments ? new AMQPTable($arguments) : [],
             $options->getTicket()
         );
@@ -148,30 +136,33 @@ protected function declareQueue(): void
     {
         $queueOptions = $this->getQueueOptions();
 
-        if (! $queueOptions || null === $queueOptions->getName()) {
+        if (! $queueOptions || '' === $queueOptions->getName()) {
             return;
         }
 
         $exchangeOptions = $this->getExchangeOptions();
         $arguments = $queueOptions->getArguments();
 
+        /** @psalm-var non-empty-list<string> $result */
         $result = $this->getChannel()->queue_declare(
             $queueOptions->getName(),
             $queueOptions->isPassive(),
             $queueOptions->isDurable(),
             $queueOptions->isExclusive(),
             $queueOptions->isAutoDelete(),
-            $queueOptions->isNoWait(),
+            false,
             $arguments ? new AMQPTable($arguments) : [],
             $queueOptions->getTicket()
         );
-        if (is_array($result)) {
-            [$queueName] = $result;
-        } else {
-            $queueName = null;
+
+        [$queueName] = $result;
+
+        if (null === $exchangeOptions) {
+            throw new \RuntimeException('Unable to create queue bindings: no exchange configuration provided');
         }
 
         $routingKeys = $queueOptions->getRoutingKeys();
+
         if (! count($routingKeys)) {
             $routingKeys = [''];
         }

src/BaseAmqp.php

@@ -5,31 +5,44 @@
 namespace RabbitMqModule;
 
 use BadFunctionCallException;
-use function count;
+use PhpAmqpLib\Channel\AMQPChannel;
+use PhpAmqpLib\Connection\AbstractConnection;
+use RabbitMqModule\Options\Queue;
 use function extension_loaded;
 use function function_exists;
-use Laminas\EventManager\EventManagerAwareInterface;
-use Laminas\EventManager\EventManagerAwareTrait;
 use PhpAmqpLib\Message\AMQPMessage;
 
-abstract class BaseConsumer extends BaseAmqp implements EventManagerAwareInterface
+/**
+ * @psalm-type ConsumerHandler = callable(AMQPMessage): (int|null|void)
+ */
+abstract class BaseConsumer extends BaseAmqp
 {
-    use EventManagerAwareTrait;
+    protected ?string $consumerTag = null;
 
-    /** @var null|string */
-    protected $consumerTag;
-
-    /** @var callable */
+    /**
+     * @psalm-var callable(AMQPMessage): (int|void)
+     * @var callable
+     */
     protected $callback;
 
-    /** @var bool */
-    protected $forceStop = false;
+    protected bool $forceStop = false;
+
+    protected int $idleTimeout = 0;
 
-    /** @var int */
-    protected $idleTimeout = 0;
+    protected bool $signalsEnabled = true;
 
-    /** @var bool */
-    protected $signalsEnabled = true;
+    protected string $queueName;
+
+    /**
+     * @psalm-param callable(AMQPMessage): (int|void) $callback
+     */
+    public function __construct(AbstractConnection $connection, Queue $queueOptions, callable $callback, AMQPChannel $channel = null)
+    {
+        parent::__construct($connection, $channel);
+        $this->setQueueOptions($queueOptions);
+        $this->queueName = $queueOptions->getName();
+        $this->callback = $callback;
+    }
 
     public function isSignalsEnabled(): bool
     {
@@ -55,11 +68,19 @@ public function setConsumerTag(string $consumerTag): void
         $this->consumerTag = $consumerTag;
     }
 
+    /**
+     * @psalm-return callable(AMQPMessage): (int|void)
+     * @return callable
+     */
     public function getCallback(): callable
     {
         return $this->callback;
     }
 
+    /**
+     * @param callable(AMQPMessage): (int|void) $callback
+     * @return void
+     */
     public function setCallback(callable $callback): void
     {
         $this->callback = $callback;
@@ -82,7 +103,7 @@ public function start(): void
     {
         $this->setupConsumer();
 
-        while (count($this->getChannel()->callbacks)) {
+        while ($this->getChannel()->is_consuming()) {
             $this->getChannel()->wait();
         }
     }
@@ -94,7 +115,7 @@ protected function setupConsumer(): void
         }
 
         $this->getChannel()->basic_consume(
-            $this->getQueueOptions()->getName(),
+            $this->queueName,
             $this->getConsumerTag(),
             false,
             false,
@@ -107,12 +128,8 @@ protected function setupConsumer(): void
     /**
      * Sets the qos settings for the current channel
      * Consider that prefetchSize and global do not work with rabbitMQ version <= 8.0.
-     *
-     * @param int  $prefetchSize
-     * @param int  $prefetchCount
-     * @param bool $global
      */
-    public function setQosOptions($prefetchSize = 0, $prefetchCount = 0, $global = false): void
+    public function setQosOptions(int $prefetchSize = 0, int $prefetchCount = 0, bool $global = false): void
     {
         $this->getChannel()->basic_qos($prefetchSize, $prefetchCount, $global);
     }

src/BaseConsumer.php

@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+namespace RabbitMqModule\Command\Factory;
+
+use Psr\Container\ContainerInterface;
+use RabbitMqModule\Command\ListConsumersCommand;
+use RabbitMqModule\ConfigProvider;
+
+/**
+ * @psalm-import-type ConfigArray from ConfigProvider
+ */
+final class ListConsumersCommandFactory
+{
+    public function __invoke(ContainerInterface $container): ListConsumersCommand
+    {
+        /** @psalm-var ConfigArray $config */
+        $config = $container->get('config');
+
+        return new ListConsumersCommand($config['rabbitmq']['consumer'] ?? []);
+    }
+}