Added amqpMessage property to events for amqp (#6518)

hyperf · Feb 2, 2024 · cb48673 · cb48673
1 parent dae5588
commit cb48673
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 3 deletions.
diff --git a/src/Consumer.php b/src/Consumer.php
@@ -171,9 +171,9 @@ protected function getCallback(ConsumerMessageInterface $consumerMessage, AMQPMe
             try {
                 $data = $consumerMessage->unserialize($message->getBody());
 
-                $this->eventDispatcher?->dispatch(new BeforeConsume($consumerMessage));
+                $this->eventDispatcher?->dispatch(new BeforeConsume($consumerMessage, $message));
                 $result = $consumerMessage->consumeMessage($data, $message);
-                $this->eventDispatcher?->dispatch(new AfterConsume($consumerMessage, $result));
+                $this->eventDispatcher?->dispatch(new AfterConsume($consumerMessage, $result, $message));
             } catch (Throwable $exception) {
                 $this->eventDispatcher?->dispatch(new FailToConsume($consumerMessage, $exception, $message));
                 if ($this->container->has(FormatterInterface::class)) {

diff --git a/src/Event/AfterConsume.php b/src/Event/AfterConsume.php
@@ -13,10 +13,11 @@
 
 use Hyperf\Amqp\Message\ConsumerMessageInterface;
 use Hyperf\Amqp\Result;
+use PhpAmqpLib\Message\AMQPMessage;
 
 class AfterConsume extends ConsumeEvent
 {
-    public function __construct(ConsumerMessageInterface $message, protected Result $result)
+    public function __construct(ConsumerMessageInterface $message, protected Result $result, protected AMQPMessage $amqpMessage)
     {
         parent::__construct($message);
     }
@@ -25,4 +26,9 @@ public function getResult(): Result
     {
         return $this->result;
     }
+
+    public function getAMQPMessage(): AMQPMessage
+    {
+        return $this->amqpMessage;
+    }
 }
diff --git a/src/Event/BeforeConsume.php b/src/Event/BeforeConsume.php
@@ -11,6 +11,18 @@
  */
 namespace Hyperf\Amqp\Event;
 
+use Hyperf\Amqp\Message\ConsumerMessageInterface;
+use PhpAmqpLib\Message\AMQPMessage;
+
 class BeforeConsume extends ConsumeEvent
 {
+    public function __construct(ConsumerMessageInterface $message, protected AMQPMessage $amqpMessage)
+    {
+        parent::__construct($message);
+    }
+
+    public function getAMQPMessage(): AMQPMessage
+    {
+        return $this->amqpMessage;
+    }
 }