Add types and support for PHP 8

Author: thomasvargiu

composer.json

@@ -25,17 +25,15 @@
         "php-amqplib/php-amqplib": "^3.0.0",
         "laminas/laminas-stdlib": "^3.3.0",
         "laminas/laminas-servicemanager": "^3.11",
-        "laminas/laminas-modulemanager": "^2.10.0",
-        "laminas/laminas-mvc": "^3.3",
-        "laminas/laminas-cli": "^1.0",
         "laminas/laminas-serializer": "^2.10.0"
     },
     "require-dev": {
         "phpunit/phpunit": "^9.3",
         "friendsofphp/php-cs-fixer": "^3.11",
         "phpspec/prophecy": "^1.12.0",
         "phpspec/prophecy-phpunit": "^2.0",
-        "vimeo/psalm": "^4.27"
+        "vimeo/psalm": "^4.27",
+        "laminas/laminas-cli": "^1.0"
     },
     "autoload": {
         "psr-4": {

composer.lock

@@ -9,13 +9,16 @@
 use PhpAmqpLib\Connection\AbstractConnection;
 use PhpAmqpLib\Wire\AMQPTable;
 use RabbitMqModule\Options\Exchange as ExchangeOptions;
+use RabbitMqModule\Options\Qos;
 use RabbitMqModule\Options\Queue as QueueOptions;
 use RabbitMqModule\Service\SetupFabricAwareInterface;
 
 abstract class BaseAmqp implements SetupFabricAwareInterface
 {
     protected AbstractConnection $connection;
 
+    protected ?Qos $qos = null;
+
     private ?AMQPChannel $channel = null;
 
     protected ?QueueOptions $queueOptions = null;
@@ -28,31 +31,65 @@ abstract class BaseAmqp implements SetupFabricAwareInterface
 
     protected bool $queueDeclared = false;
 
-    public function __construct(AbstractConnection $connection, AMQPChannel $channel = null)
+    public function __construct(AbstractConnection $connection)
     {
         $this->connection = $connection;
-        $this->channel = $channel;
     }
 
+    /**
+     * @internal
+     *
+     * @psalm-internal RabbitMqModule
+     */
     public function getConnection(): AbstractConnection
     {
         return $this->connection;
     }
 
+    /**
+     * @internal
+     *
+     * @psalm-internal RabbitMqModule
+     */
     public function getChannel(): AMQPChannel
     {
-        if (! $this->channel) {
-            $this->channel = $this->getConnection()->channel();
+        if ($this->channel) {
+            return $this->channel;
+        }
+
+        $this->channel = $this->getConnection()->channel();
+
+        if ($this->qos) {
+            $this->channel->basic_qos(
+                $this->qos->getPrefetchSize(),
+                $this->qos->getPrefetchCount(),
+                false
+            );
         }
 
         return $this->channel;
     }
 
+    public function setQos(?Qos $qos): void
+    {
+        $this->qos = $qos;
+    }
+
+    /**
+     * @internal
+     *
+     * @psalm-internal RabbitMqModule
+     */
     public function setChannel(AMQPChannel $channel): void
     {
         $this->channel = $channel;
     }
 
+    /**
+     * @internal
+     *
+     * @psalm-internal RabbitMqModule
+     */
     public function getQueueOptions(): ?QueueOptions
     {
         return $this->queueOptions;
@@ -63,6 +100,11 @@ public function setQueueOptions(?QueueOptions $queueOptions): void
         $this->queueOptions = $queueOptions;
     }
 
+    /**
+     * @internal
+     *
+     * @psalm-internal RabbitMqModule
+     */
     public function getExchangeOptions(): ?ExchangeOptions
     {
         return $this->exchangeOptions;
@@ -73,6 +115,11 @@ public function setExchangeOptions(?ExchangeOptions $exchangeOptions): void
         $this->exchangeOptions = $exchangeOptions;
     }
 
+    /**
+     * @internal
+     *
+     * @psalm-internal RabbitMqModule
+     */
     public function isAutoSetupFabricEnabled(): bool
     {
         return $this->autoSetupFabricEnabled;
@@ -195,6 +242,10 @@ public function setupFabric(): void
 
     /**
      * Reconnect
+     *
+     * @internal
+     *
+     * @psalm-internal RabbitMqModule
      */
     public function reconnect(): void
     {

src/BaseAmqp.php

@@ -5,12 +5,11 @@
 namespace RabbitMqModule;
 
 use BadFunctionCallException;
-use PhpAmqpLib\Channel\AMQPChannel;
-use PhpAmqpLib\Connection\AbstractConnection;
-use RabbitMqModule\Options\Queue;
 use function extension_loaded;
 use function function_exists;
+use PhpAmqpLib\Connection\AbstractConnection;
 use PhpAmqpLib\Message\AMQPMessage;
+use RabbitMqModule\Options\Queue;
 
 /**
  * @psalm-type ConsumerHandler = callable(AMQPMessage): (int|null|void)
@@ -21,6 +20,7 @@ abstract class BaseConsumer extends BaseAmqp
 
     /**
      * @psalm-var callable(AMQPMessage): (int|void)
+     *
      * @var callable
      */
     protected $callback;
@@ -36,14 +36,19 @@ abstract class BaseConsumer extends BaseAmqp
     /**
      * @psalm-param callable(AMQPMessage): (int|void) $callback
      */
-    public function __construct(AbstractConnection $connection, Queue $queueOptions, callable $callback, AMQPChannel $channel = null)
+    public function __construct(AbstractConnection $connection, Queue $queueOptions, callable $callback)
     {
-        parent::__construct($connection, $channel);
+        parent::__construct($connection);
         $this->setQueueOptions($queueOptions);
         $this->queueName = $queueOptions->getName();
         $this->callback = $callback;
     }
 
+    /**
+     * @internal
+     *
+     * @psalm-internal RabbitMqModule
+     */
     public function isSignalsEnabled(): bool
     {
         return $this->signalsEnabled;
@@ -69,7 +74,12 @@ public function setConsumerTag(string $consumerTag): void
     }
 
     /**
+     * @internal
+     *
+     * @psalm-internal RabbitMqModule
+     *
      * @psalm-return callable(AMQPMessage): (int|void)
+     *
      * @return callable
      */
     public function getCallback(): callable
@@ -79,13 +89,19 @@ public function getCallback(): callable
 
     /**
      * @param callable(AMQPMessage): (int|void) $callback
+     *
      * @return void
      */
     public function setCallback(callable $callback): void
     {
         $this->callback = $callback;
     }
 
+    /**
+     * @internal
+     *
+     * @psalm-internal RabbitMqModule
+     */
     public function getIdleTimeout(): int
     {
         return $this->idleTimeout;
@@ -128,6 +144,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.
+     *
+     * @deprecated
      */
     public function setQosOptions(int $prefetchSize = 0, int $prefetchCount = 0, bool $global = false): void
     {

src/BaseConsumer.php

@@ -56,6 +56,8 @@ protected function execute(InputInterface $input, OutputInterface $output): int
      * @throws NotFoundExceptionInterface
      *
      * @psalm-return list<mixed>
+     *
+     * @psalm-suppress MixedAssignment
      */
     private function getServiceParts(): array
     {
@@ -78,7 +80,6 @@ private function getServiceParts(): array
 
             $keys = array_keys($config['rabbitmq'][$serviceKey]);
             foreach ($keys as $key) {
-                /** @psalm-suppress MixedAssignment */
                 $parts[] = $this->container->get(sprintf('rabbitmq.%s.%s', $serviceKey, $key));
             }
         }

src/Command/SetupFabricCommand.php

@@ -17,7 +17,7 @@
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 
-final class StartConsumerCommand extends Command
+class StartConsumerCommand extends Command
 {
     public const NAME = 'rabbitmq:consumers:start';
 
@@ -83,8 +83,8 @@ protected function execute(InputInterface $input, OutputInterface $output): int
                 );
             }
 
-            pcntl_signal(SIGTERM, [$this, 'stopConsumer']);
-            pcntl_signal(SIGINT, [$this, 'stopConsumer']);
+            pcntl_signal(SIGTERM, fn () => $this->stopConsumer());
+            pcntl_signal(SIGINT, fn () => $this->stopConsumer());
         }
         // @codeCoverageIgnoreEnd
 
@@ -93,7 +93,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
         return Command::SUCCESS;
     }
 
-    public function stopConsumer(): void
+    protected function stopConsumer(): void
     {
         if ($this->consumer instanceof Consumer) {
             $this->consumer->forceStopConsumer();

src/Command/StartConsumerCommand.php

@@ -1,11 +1,10 @@
 <?php
+
 declare(strict_types=1);
 
 namespace RabbitMqModule;
 
 use Laminas\ServiceManager\Factory\InvokableFactory;
-use RabbitMqModule\Command;
-use RabbitMqModule\Service;
 use RabbitMqModule\Service\AbstractFactory;
 
 /**
@@ -18,6 +17,7 @@
  * @psalm-import-type ConsumerOptions from Options\Consumer
  * @psalm-import-type RpcClientOptions from Options\RpcClient
  * @psalm-import-type RpcServerOptions from Options\RpcServer
+ *
  * @psalm-type ConfigArray = array{
  *   rabbitmq_factories: array{
  *     connection: class-string<AbstractFactory>,
@@ -40,6 +40,7 @@ final class ConfigProvider
 {
     /**
      * @psalm-return ConfigArray
+     *
      * @return array<string, mixed>
      */
     public function __invoke(): array
@@ -48,12 +49,12 @@ public function __invoke(): array
             'dependencies' => $this->getDependencies(),
             'rabbitmq' => [
                 'connection' => [
-                    'default' => []
+                    'default' => [],
                 ],
                 'producer' => [],
                 'consumer' => [],
                 'rpc_server' => [],
-                'rpc_client' => []
+                'rpc_client' => [],
             ],
             'rabbitmq_factories' => [
                 'connection' => Service\ConnectionFactory::class,
@@ -69,8 +70,8 @@ public function __invoke(): array
                     Command\StartRpcServerCommand::NAME => Command\StartRpcServerCommand::class,
                     Command\PublishMessageCommand::NAME => Command\PublishMessageCommand::class,
                     Command\SetupFabricCommand::NAME => Command\SetupFabricCommand::class,
-                ]
-            ]
+                ],
+            ],
         ];
     }
 

src/ConfigProvider.php

@@ -49,6 +49,11 @@ public function consume(): void
         }
     }
 
+    /**
+     * @internal
+     *
+     * @psalm-internal RabbitMqModule
+     */
     public function processMessage(AMQPMessage $message): void
     {
         $processFlag = call_user_func($this->getCallback(), $message);
@@ -60,11 +65,12 @@ public function processMessage(AMQPMessage $message): void
      */
     protected function handleProcessMessage(AMQPMessage $msg, $processFlag): void
     {
-        /** @psalm-suppress DocblockTypeContradiction */
+        /* @psalm-suppress DocblockTypeContradiction */
         if (null !== $processFlag && ! is_int($processFlag)) {
             trigger_error(
                 'Consumer handler should return an integer or void/null. Returning a different type is deprecated',
-            E_USER_DEPRECATED);
+                E_USER_DEPRECATED
+            );
         }
 
         $channel = $msg->getChannel() ?: $this->getChannel();

src/Consumer.php

@@ -14,33 +14,33 @@ interface ConsumerInterface
     /**
      * Flag for message ack.
      *
-     * @deprecated Use {@see Consumer::MSG_ACK} instead.
+     * @deprecated use {@see Consumer::MSG_ACK} instead
      */
     public const MSG_ACK = Consumer::MSG_ACK;
 
     /**
      * Flag single for message nack and requeue.
      *
-     * @deprecated Use {@see Consumer::MSG_SINGLE_NACK_REQUEUE} instead.
+     * @deprecated use {@see Consumer::MSG_SINGLE_NACK_REQUEUE} instead
      */
     public const MSG_SINGLE_NACK_REQUEUE = Consumer::MSG_SINGLE_NACK_REQUEUE;
 
     /**
      * Flag for reject and requeue.
      *
-     * @deprecated Use {@see Consumer::MSG_REJECT_REQUEUE} instead.
+     * @deprecated use {@see Consumer::MSG_REJECT_REQUEUE} instead
      */
     public const MSG_REJECT_REQUEUE = Consumer::MSG_REJECT_REQUEUE;
 
     /**
      * Flag for reject and drop.
      *
-     * @deprecated Use {@see Consumer::MSG_REJECT} instead.
+     * @deprecated use {@see Consumer::MSG_REJECT} instead
      */
     public const MSG_REJECT = Consumer::MSG_REJECT;
 
     /**
-     * @deprecated Use an invokable class instead.
+     * @deprecated use an invokable class instead
      */
     public function execute(AMQPMessage $message): ?int;
 }