wip

src/Consumers/Consumer.php

@@ -88,6 +88,11 @@ public function __construct(
         $this->identifier = $identifier ?? str()->random();
     }
 
+    public function identifier(): string
+    {
+        return $this->identifier;
+    }
+
     /**
      * Consume messages from a kafka topic in loop.
      *
@@ -145,7 +150,7 @@ public function consume(): void
             ));
 
             $callback = $this->whenStopConsuming;
-            $callback(...)();
+            $callback(...)($this);
         }
 
         $this->dispatcher->dispatch(new ConsumerStopped(
@@ -255,7 +260,7 @@ private function executeMessage(Message $message): void
             $success = true;
 
             // Dispatch an event informing that a message was consumed.
-            $this->dispatcher->dispatch(new MessageConsumed($consumedMessage));
+            $this->dispatcher->dispatch(new MessageConsumed($consumedMessage, $this->identifier()));
         } catch (Throwable $throwable) {
             $this->logger->error($message, $throwable);
             $success = $this->handleException($throwable, $message);

src/Contracts/MessageConsumer.php

@@ -11,6 +11,8 @@ interface MessageConsumer
      */
     public function consume(): void;
 
+    public function identifier(): string;
+
     /** Requests the consumer to stop after it's finished processing any messages to allow graceful exit. */
     public function stopConsuming(): void;
 

src/Events/MessageConsumed.php

@@ -7,7 +7,8 @@
 final class MessageConsumed
 {
     public function __construct(
-        public readonly ConsumerMessage $message
+        public readonly ConsumerMessage $message,
+        public readonly string $consumerIdentifier,
     ) {
     }