Optimized code for hyperf/amqp. (#6045)

Co-authored-by: 李铭昕 <[email protected]>
hyperf · Aug 21, 2023 · ad457f5 · ad457f5
1 parent 7311f48
commit ad457f5
Show file tree

Hide file tree

Showing 12 changed files with 60 additions and 56 deletions.
diff --git a/src/AMQPConnection.php b/src/AMQPConnection.php
@@ -79,7 +79,7 @@ public function __construct(
         $this->confirmPool = new Channel(static::CONFIRM_CHANNEL_POOL_LENGTH);
     }
 
-    public function write($data)
+    public function write($data): void
     {
         $this->loop();
 
@@ -105,7 +105,7 @@ public function setParams(Params $params): static
         return $this;
     }
 
-    public function getIO()
+    public function getIO(): AbstractIO
     {
         return $this->io;
     }
@@ -124,7 +124,7 @@ public function getChannel(): AMQPChannel
         return $this->channel($id);
     }
 
-    public function channel($channel_id = null)
+    public function channel($channel_id = null): AMQPChannel
     {
         $this->channelManager->close($channel_id);
         $this->channelManager->get($channel_id, true);

diff --git a/src/ConnectionFactory.php b/src/ConnectionFactory.php
@@ -29,14 +29,14 @@ class ConnectionFactory
     /**
      * @var AMQPConnection[][]
      */
-    protected $connections = [];
+    protected array $connections = [];
 
     public function __construct(protected ContainerInterface $container)
     {
         $this->config = $this->container->get(ConfigInterface::class);
     }
 
-    public function refresh(string $pool)
+    public function refresh(string $pool): void
     {
         $config = $this->getConfig($pool);
         $count = $config['pool']['connections'] ?? 1;

diff --git a/src/ConsumerFactory.php b/src/ConsumerFactory.php
@@ -16,7 +16,7 @@
 
 class ConsumerFactory
 {
-    public function __invoke(ContainerInterface $container)
+    public function __invoke(ContainerInterface $container): Consumer
     {
         return new Consumer(
             $container,

diff --git a/src/ConsumerManager.php b/src/ConsumerManager.php
@@ -26,7 +26,7 @@ public function __construct(private ContainerInterface $container)
     {
     }
 
-    public function run()
+    public function run(): void
     {
         $classes = AnnotationCollector::getClassesByAnnotation(ConsumerAnnotation::class);
         /**
@@ -57,15 +57,9 @@ public function run()
     private function createProcess(ConsumerMessageInterface $consumerMessage): AbstractProcess
     {
         return new class($this->container, $consumerMessage) extends AbstractProcess {
-            /**
-             * @var \Hyperf\Amqp\Consumer
-             */
-            private $consumer;
+            private Consumer $consumer;
 
-            /**
-             * @var ConsumerMessageInterface
-             */
-            private $consumerMessage;
+            private ConsumerMessageInterface $consumerMessage;
 
             public function __construct(ContainerInterface $container, ConsumerMessageInterface $consumerMessage)
             {

diff --git a/src/IO/SwooleIO.php b/src/IO/SwooleIO.php
@@ -13,6 +13,7 @@
 
 use InvalidArgumentException;
 use PhpAmqpLib\Exception\AMQPConnectionClosedException;
+use PhpAmqpLib\Exception\AMQPIOException;
 use PhpAmqpLib\Exception\AMQPRuntimeException;
 use PhpAmqpLib\Wire\AMQPWriter;
 use PhpAmqpLib\Wire\IO\AbstractIO;
@@ -58,12 +59,12 @@ public function __construct(
      *
      * @throws AMQPRuntimeException
      */
-    public function connect()
+    public function connect(): void
     {
         $this->sock = $this->makeClient();
     }
 
-    public function read($len)
+    public function read($len): string
     {
         $data = $this->sock->recvAll($len, $this->readWriteTimeout);
         if ($data === false || strlen($data) !== $len) {
@@ -73,7 +74,7 @@ public function read($len)
         return $data;
     }
 
-    public function write($data)
+    public function write($data): void
     {
         $len = $this->sock->sendAll($data, $this->readWriteTimeout);
 
@@ -87,27 +88,27 @@ public function check_heartbeat()
     {
     }
 
-    public function close()
+    public function close(): void
     {
         $this->sock && $this->sock->close();
     }
 
-    public function select(?int $sec, int $usec = 0)
+    public function select(?int $sec, int $usec = 0): int
     {
         return 1;
     }
 
-    public function disableHeartbeat()
+    public function disableHeartbeat(): AbstractIO
     {
         return $this;
     }
 
-    public function reenableHeartbeat()
+    public function reenableHeartbeat(): AbstractIO
     {
         return $this;
     }
 
-    protected function makeClient()
+    protected function makeClient(): Socket
     {
         $sock = new Socket(AF_INET, SOCK_STREAM, 0);
 
@@ -124,7 +125,10 @@ protected function makeClient()
         return $sock;
     }
 
-    protected function write_heartbeat()
+    /**
+     * @throws AMQPIOException
+     */
+    protected function write_heartbeat(): void
     {
         $pkt = new AMQPWriter();
         $pkt->write_octet(8);
@@ -134,7 +138,7 @@ protected function write_heartbeat()
         $this->write($pkt->getvalue());
     }
 
-    protected function do_select($sec, $usec)
+    protected function do_select($sec, $usec): int
     {
         return 1;
     }

diff --git a/src/IO/SwowIO.php b/src/IO/SwowIO.php
@@ -59,12 +59,12 @@ public function __construct(
      *
      * @throws AMQPRuntimeException
      */
-    public function connect()
+    public function connect(): void
     {
         $this->sock = $this->makeClient();
     }
 
-    public function read($len)
+    public function read($len): string
     {
         $data = $this->sock->recvAll($len, $this->readWriteTimeout);
         if ($data === false || strlen($data) !== $len) {
@@ -74,7 +74,7 @@ public function read($len)
         return $data;
     }
 
-    public function write($data)
+    public function write($data): void
     {
         $len = $this->sock->sendAll($data, $this->readWriteTimeout);
 
@@ -88,27 +88,27 @@ public function check_heartbeat()
     {
     }
 
-    public function close()
+    public function close(): void
     {
         $this->sock && $this->sock->close();
     }
 
-    public function select(?int $sec, int $usec = 0)
+    public function select(?int $sec, int $usec = 0): int
     {
         return 1;
     }
 
-    public function disableHeartbeat()
+    public function disableHeartbeat(): AbstractIO
     {
         return $this;
     }
 
-    public function reenableHeartbeat()
+    public function reenableHeartbeat(): AbstractIO
     {
         return $this;
     }
 
-    protected function makeClient()
+    protected function makeClient(): Socket
     {
         $sock = new Socket(Socket::TYPE_TCP);
 
@@ -125,7 +125,7 @@ protected function makeClient()
         return $sock;
     }
 
-    protected function write_heartbeat()
+    protected function write_heartbeat(): void
     {
         $pkt = new AMQPWriter();
         $pkt->write_octet(8);
@@ -135,7 +135,7 @@ protected function write_heartbeat()
         $this->write($pkt->getvalue());
     }
 
-    protected function do_select($sec, $usec)
+    protected function do_select($sec, $usec): int
     {
         return 1;
     }

diff --git a/src/Message/ConsumerMessage.php b/src/Message/ConsumerMessage.php
@@ -147,7 +147,7 @@ public function getContainer(): ?ContainerInterface
         return $this->container;
     }
 
-    protected function reply($data, AMQPMessage $message)
+    protected function reply(mixed $data, AMQPMessage $message): void
     {
         $packer = ApplicationContext::getContainer()->get(Packer::class);
 

diff --git a/src/Message/ProducerDelayedMessageTrait.php b/src/Message/ProducerDelayedMessageTrait.php
@@ -25,7 +25,7 @@ trait ProducerDelayedMessageTrait
      * Set the delay time.
      * @return $this
      */
-    public function setDelayMs(int $millisecond, string $name = 'x-delay'): self
+    public function setDelayMs(int $millisecond, string $name = 'x-delay'): static
     {
         $this->properties['application_headers'] = new AMQPTable([$name => $millisecond]);
         return $this;

diff --git a/src/Params.php b/src/Params.php
@@ -116,64 +116,73 @@ public function isCloseOnDestruct(): bool
         return $this->closeOnDestruct;
     }
 
-    public function setCloseOnDestruct(bool $closeOnDestruct)
+    public function setCloseOnDestruct(bool $closeOnDestruct): static
     {
         $this->closeOnDestruct = $closeOnDestruct;
         return $this;
     }
 
-    public function setInsist(bool $insist)
+    public function setInsist(bool $insist): static
     {
         $this->insist = $insist;
+        return $this;
     }
 
-    public function setLoginMethod(string $loginMethod)
+    public function setLoginMethod(string $loginMethod): static
     {
         $this->loginMethod = $loginMethod;
+        return $this;
     }
 
-    public function setLocale(string $locale)
+    public function setLocale(string $locale): static
     {
         $this->locale = $locale;
+        return $this;
     }
 
-    public function setConnectionTimeout(int $connectionTimeout)
+    public function setConnectionTimeout(int $connectionTimeout): static
     {
         $this->connectionTimeout = $connectionTimeout;
+        return $this;
     }
 
-    public function setReadWriteTimeout(int $readWriteTimeout)
+    public function setReadWriteTimeout(int $readWriteTimeout): static
     {
         $this->readWriteTimeout = $readWriteTimeout;
+        return $this;
     }
 
-    public function setKeepalive(bool $keepalive)
+    public function setKeepalive(bool $keepalive): static
     {
         $this->keepalive = $keepalive;
+        return $this;
     }
 
-    public function setHeartbeat(int $heartbeat)
+    public function setHeartbeat(int $heartbeat): static
     {
         $this->heartbeat = $heartbeat;
+        return $this;
     }
 
     public function getChannelRpcTimeout(): float
     {
         return $this->channelRpcTimeout;
     }
 
-    public function setChannelRpcTimeout(float $channelRpcTimeout)
+    public function setChannelRpcTimeout(float $channelRpcTimeout): static
     {
         $this->channelRpcTimeout = $channelRpcTimeout;
+        return $this;
     }
 
     public function getMaxIdleChannels(): int
     {
         return $this->maxIdleChannels;
     }
 
-    public function setMaxIdleChannels(int $maxIdleChannels)
+    public function setMaxIdleChannels(int $maxIdleChannels): static
     {
         $this->maxIdleChannels = $maxIdleChannels;
+        return $this;
     }
 }
diff --git a/src/Producer.php b/src/Producer.php
@@ -25,7 +25,7 @@ public function produce(ProducerMessageInterface $producerMessage, bool $confirm
         return retry(1, fn () => $this->produceMessage($producerMessage, $confirm, $timeout));
     }
 
-    private function produceMessage(ProducerMessageInterface $producerMessage, bool $confirm = false, int $timeout = 5)
+    private function produceMessage(ProducerMessageInterface $producerMessage, bool $confirm = false, int $timeout = 5): bool
     {
         $result = false;
 
@@ -61,7 +61,7 @@ private function produceMessage(ProducerMessageInterface $producerMessage, bool
         return $result;
     }
 
-    private function injectMessageProperty(ProducerMessageInterface $producerMessage)
+    private function injectMessageProperty(ProducerMessageInterface $producerMessage): void
     {
         if (class_exists(AnnotationCollector::class)) {
             /** @var null|\Hyperf\Amqp\Annotation\Producer $annotation */

diff --git a/src/RpcChannel.php b/src/RpcChannel.php
@@ -61,17 +61,14 @@ public function setQueue(?string $queue): static
         return $this;
     }
 
-    /**
-     * @return AMQPMessage|false
-     */
-    public function wait(int $timeout)
+    public function wait(int $timeout): bool|AMQPMessage
     {
         $this->channel->wait(null, false, $timeout);
 
         return $this->chan->pop(0.001);
     }
 
-    public function close()
+    public function close(): void
     {
         $this->chan?->close();
         $this->channel->close();

diff --git a/src/RpcClient.php b/src/RpcClient.php
@@ -21,7 +21,7 @@
 
 class RpcClient extends Builder
 {
-    protected $poolChannels = [];
+    protected array $poolChannels = [];
 
     public function __construct(ContainerInterface $container, ConnectionFactory $factory, protected int $maxChannels = 64)
     {
@@ -75,7 +75,7 @@ public function call(RpcMessageInterface $rpcMessage, int $timeout = 5)
         return $result;
     }
 
-    protected function initChannel(RpcChannel $channel, QueueBuilder $builder)
+    protected function initChannel(RpcChannel $channel, QueueBuilder $builder): void
     {
         [$queue] = $channel->getChannel()->queue_declare(
             $builder->getQueue(),