Improve amqp, use methods instead of $delivery_info and optimize `B…

…eforeConsume` event (#6239)
hyperf · Oct 30, 2023 · 08c9f4d · 08c9f4d
1 parent 58b406a
commit 08c9f4d
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 9 deletions.
diff --git a/src/Consumer.php b/src/Consumer.php
@@ -165,17 +165,17 @@ protected function getConcurrent(string $pool): ?Concurrent
     protected function getCallback(ConsumerMessageInterface $consumerMessage, AMQPMessage $message): callable
     {
         return function () use ($consumerMessage, $message) {
-            $data = $consumerMessage->unserialize($message->getBody());
-            /** @var AMQPChannel $channel */
-            $channel = $message->delivery_info['channel'];
-            $deliveryTag = $message->delivery_info['delivery_tag'];
+            $channel = $message->getChannel();
+            $deliveryTag = $message->getDeliveryTag();
 
             try {
+                $data = $consumerMessage->unserialize($message->getBody());
+
                 $this->eventDispatcher?->dispatch(new BeforeConsume($consumerMessage));
                 $result = $consumerMessage->consumeMessage($data, $message);
                 $this->eventDispatcher?->dispatch(new AfterConsume($consumerMessage, $result));
             } catch (Throwable $exception) {
-                $this->eventDispatcher?->dispatch(new FailToConsume($consumerMessage, $exception));
+                $this->eventDispatcher?->dispatch(new FailToConsume($consumerMessage, $exception, $message));
                 if ($this->container->has(FormatterInterface::class)) {
                     $formatter = $this->container->get(FormatterInterface::class);
                     $this->logger->error($formatter->format($exception));

diff --git a/src/Event/FailToConsume.php b/src/Event/FailToConsume.php
@@ -12,11 +12,12 @@
 namespace Hyperf\Amqp\Event;
 
 use Hyperf\Amqp\Message\ConsumerMessageInterface;
+use PhpAmqpLib\Message\AMQPMessage;
 use Throwable;
 
 class FailToConsume extends ConsumeEvent
 {
-    public function __construct(ConsumerMessageInterface $message, protected Throwable $throwable)
+    public function __construct(ConsumerMessageInterface $message, protected Throwable $throwable, protected AMQPMessage $amqpMessage)
     {
         parent::__construct($message);
     }
@@ -25,4 +26,9 @@ public function getThrowable(): Throwable
     {
         return $this->throwable;
     }
+
+    public function getAMQPMessage(): AMQPMessage
+    {
+        return $this->amqpMessage;
+    }
 }
diff --git a/src/Message/ConsumerMessage.php b/src/Message/ConsumerMessage.php
@@ -15,7 +15,6 @@
 use Hyperf\Amqp\Packer\Packer;
 use Hyperf\Amqp\Result;
 use Hyperf\Context\ApplicationContext;
-use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Message\AMQPMessage;
 use Psr\Container\ContainerInterface;
 
@@ -151,8 +150,7 @@ protected function reply($data, AMQPMessage $message)
     {
         $packer = ApplicationContext::getContainer()->get(Packer::class);
 
-        /** @var AMQPChannel $channel */
-        $channel = $message->delivery_info['channel'];
+        $channel = $message->getChannel();
         $channel->basic_publish(
             new AMQPMessage($packer->pack($data), [
                 'correlation_id' => $message->get('correlation_id'),