[add] Добавлен интерфейс ConsumerMessageInterface

phpstan.neon.dist

@@ -5,4 +5,3 @@ parameters:
     level: 5
     paths:
         - src
-    tmpDir: build/phpstan

src/Bus.php

@@ -6,7 +6,7 @@
 use Micromus\KafkaBus\Bus\ThreadRegistry;
 use Micromus\KafkaBus\Interfaces\Bus\BusInterface;
 use Micromus\KafkaBus\Interfaces\Bus\ThreadInterface;
-use Micromus\KafkaBus\Interfaces\Messages\MessageInterface;
+use Micromus\KafkaBus\Interfaces\Producers\Messages\ProducerMessageInterface;
 
 class Bus implements BusInterface
 {
@@ -25,7 +25,7 @@ public function onConnection(string $connectionName): ThreadInterface
             ->thread($connectionName);
     }
 
-    public function publish(MessageInterface $message): void
+    public function publish(ProducerMessageInterface $message): void
     {
         $this->thread->publish($message);
     }

src/Bus/Listeners/Workers/Route.php

@@ -2,8 +2,6 @@
 
 namespace Micromus\KafkaBus\Bus\Listeners\Workers;
 
-use Micromus\KafkaBus\Messages\NativeMessageFactory;
-
 readonly class Route
 {
     public function __construct(

src/Bus/Thread.php

@@ -8,7 +8,7 @@
 use Micromus\KafkaBus\Bus\Publishers\PublisherFactory;
 use Micromus\KafkaBus\Interfaces\Bus\ThreadInterface;
 use Micromus\KafkaBus\Interfaces\Connections\ConnectionInterface;
-use Micromus\KafkaBus\Interfaces\Messages\MessageInterface;
+use Micromus\KafkaBus\Interfaces\Producers\Messages\ProducerMessageInterface;
 
 class Thread implements ThreadInterface
 {
@@ -31,7 +31,7 @@ private function getPublisher(): Publisher
         return $this->publisher;
     }
 
-    public function publish(MessageInterface $message): void
+    public function publish(ProducerMessageInterface $message): void
     {
         $this->publishMany([$message]);
     }

src/Consumers/Commiters/CommiterInterface.php

@@ -2,9 +2,9 @@
 
 namespace Micromus\KafkaBus\Consumers\Commiters;
 
-use Micromus\KafkaBus\Consumers\Messages\ConsumerMessage;
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageInterface;
 
 interface CommiterInterface
 {
-    public function commit(ConsumerMessage $consumerMessage): void;
+    public function commit(ConsumerMessageInterface $consumerMessage): void;
 }

src/Consumers/Commiters/DefaultCommiter.php

@@ -2,7 +2,7 @@
 
 namespace Micromus\KafkaBus\Consumers\Commiters;
 
-use Micromus\KafkaBus\Consumers\Messages\ConsumerMessage;
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageInterface;
 use RdKafka\KafkaConsumer;
 
 class DefaultCommiter implements CommiterInterface
@@ -12,9 +12,9 @@ public function __construct(
     ) {
     }
 
-    public function commit(ConsumerMessage $consumerMessage): void
+    public function commit(ConsumerMessageInterface $consumerMessage): void
     {
         $this->consumer
-            ->commitAsync($consumerMessage->meta->message);
+            ->commitAsync($consumerMessage->original());
     }
 }

src/Consumers/Commiters/VoidCommiter.php

@@ -2,11 +2,11 @@
 
 namespace Micromus\KafkaBus\Consumers\Commiters;
 
-use Micromus\KafkaBus\Consumers\Messages\ConsumerMessage;
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageInterface;
 
 class VoidCommiter implements CommiterInterface
 {
-    public function commit(ConsumerMessage $consumerMessage): void
+    public function commit(ConsumerMessageInterface $consumerMessage): void
     {
     }
 }

src/Consumers/Consumer.php

@@ -3,11 +3,11 @@
 namespace Micromus\KafkaBus\Consumers;
 
 use Micromus\KafkaBus\Consumers\Commiters\CommiterInterface;
-use Micromus\KafkaBus\Consumers\Messages\ConsumerMessage;
 use Micromus\KafkaBus\Consumers\Messages\ConsumerMessageConverter;
 use Micromus\KafkaBus\Interfaces\Consumers\ConsumerInterface;
 use Micromus\KafkaBus\Exceptions\Consumers\ConsumerException;
 use Micromus\KafkaBus\Exceptions\Consumers\MessageConsumerException;
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageInterface;
 use Micromus\KafkaBus\Support\RetryRepeater;
 use RdKafka\Exception;
 use RdKafka\KafkaConsumer;
@@ -33,7 +33,7 @@ public function __destruct()
         $this->consumer->close();
     }
 
-    public function getMessage(): ConsumerMessage
+    public function getMessage(): ConsumerMessageInterface
     {
         try {
             $message = $this->consumer
@@ -51,7 +51,7 @@ public function getMessage(): ConsumerMessage
         }
     }
 
-    public function commit(ConsumerMessage $consumerMessage): void
+    public function commit(ConsumerMessageInterface $consumerMessage): void
     {
         $this->retryRepeater
             ->execute(fn () => $this->commiter->commit($consumerMessage));

src/Consumers/ConsumerStream.php

@@ -2,12 +2,12 @@
 
 namespace Micromus\KafkaBus\Consumers;
 
-use Micromus\KafkaBus\Consumers\Messages\ConsumerMessage;
 use Micromus\KafkaBus\Exceptions\Consumers\MessageConsumerNotHandledException;
 use Micromus\KafkaBus\Interfaces\Consumers\ConsumerInterface;
 use Micromus\KafkaBus\Interfaces\Consumers\ConsumerStreamInterface;
 use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageHandlerInterface;
 use Micromus\KafkaBus\Exceptions\Consumers\MessageConsumerException;
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageInterface;
 use Micromus\KafkaBus\Testing\Exceptions\KafkaMessagesEndedException;
 
 class ConsumerStream implements ConsumerStreamInterface
@@ -49,12 +49,12 @@ public function listen(): void
     }
 
     /**
-     * @param ConsumerMessage $message
+     * @param ConsumerMessageInterface $message
      * @return void
      *
      * @throws MessageConsumerNotHandledException
      */
-    private function handleMessage(ConsumerMessage $message): void
+    private function handleMessage(ConsumerMessageInterface $message): void
     {
         $this->consumerMessageHandler->handle($message);
         $this->consumer->commit($message);

src/Consumers/Messages/ConsumerMessage.php

@@ -2,27 +2,43 @@
 
 namespace Micromus\KafkaBus\Consumers\Messages;
 
-class ConsumerMessage
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageInterface;
+use RdKafka\Message;
+
+final class ConsumerMessage implements ConsumerMessageInterface
 {
     public function __construct(
-        public string $payload,
-        public array $headers,
-        public readonly ConsumerMeta $meta
+        protected Message $message
     ) {
     }
 
+    public function msgId(): string
+    {
+        return "{$this->message->partition}-{$this->message->offset}";
+    }
+
     public function topicName(): string
     {
-        return $this->meta->message->topic_name;
+        return $this->message->topic_name;
     }
 
     public function key(): ?string
     {
-        return $this->meta->message->key;
+        return $this->message->key;
     }
 
-    public function msgId(): string
+    public function payload(): string
+    {
+        return $this->message->payload;
+    }
+
+    public function headers(): array
+    {
+        return $this->message->headers;
+    }
+
+    public function original(): Message
     {
-        return "{$this->meta->message->partition}-{$this->meta->message->offset}";
+        return $this->message;
     }
 }

src/Consumers/Messages/ConsumerMessageConverter.php

@@ -2,16 +2,13 @@
 
 namespace Micromus\KafkaBus\Consumers\Messages;
 
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageInterface;
 use RdKafka\Message;
 
 class ConsumerMessageConverter
 {
-    public function fromKafka(Message $message): ConsumerMessage
+    public function fromKafka(Message $message): ConsumerMessageInterface
     {
-        return new ConsumerMessage(
-            $message->payload,
-            $message->headers,
-            new ConsumerMeta($message)
-        );
+        return new ConsumerMessage($message);
     }
 }

src/Consumers/Messages/ConsumerMessageHandler.php

@@ -5,6 +5,7 @@
 use Micromus\KafkaBus\Consumers\Router\ConsumerRouter;
 use Micromus\KafkaBus\Exceptions\Consumers\MessageConsumerNotHandledException;
 use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageHandlerInterface;
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageInterface;
 use Micromus\KafkaBus\Interfaces\Messages\MessagePipelineInterface;
 use Throwable;
 
@@ -21,7 +22,7 @@ public function topics(): array
         return $this->consumerRouter->topics();
     }
 
-    public function handle(ConsumerMessage $message): void
+    public function handle(ConsumerMessageInterface $message): void
     {
         $this->messagePipeline
             ->then($message, $this->handleMessage(...));
@@ -30,7 +31,7 @@ public function handle(ConsumerMessage $message): void
     /**
      * @throws MessageConsumerNotHandledException
      */
-    protected function handleMessage(ConsumerMessage $message): ConsumerMessage
+    protected function handleMessage(ConsumerMessageInterface $message): ConsumerMessageInterface
     {
         try {
             $this->consumerRouter

src/Consumers/Messages/NativeMessageFactory.php

@@ -0,0 +1,14 @@
+<?php
+
+namespace Micromus\KafkaBus\Consumers\Messages;
+
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageInterface;
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\MessageFactoryInterface;
+
+class NativeMessageFactory implements MessageFactoryInterface
+{
+    public function fromKafka(ConsumerMessageInterface $message): ConsumerMessageInterface
+    {
+        return $message;
+    }
+}

src/Consumers/Router/ConsumerRouter.php

@@ -2,7 +2,7 @@
 
 namespace Micromus\KafkaBus\Consumers\Router;
 
-use Micromus\KafkaBus\Consumers\Messages\ConsumerMessage;
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageInterface;
 use Micromus\KafkaBus\Interfaces\ResolverInterface;
 use Micromus\KafkaBus\Exceptions\Consumers\RouteConsumerException;
 
@@ -24,7 +24,7 @@ public function topics(): array
         return $this->routes->topics();
     }
 
-    public function handle(ConsumerMessage $consumerMessage): void
+    public function handle(ConsumerMessageInterface $consumerMessage): void
     {
         $executor = $this->getOrCreateExecutor($consumerMessage->topicName());
         $executor->execute($consumerMessage);

src/Consumers/Router/Executor.php

@@ -2,8 +2,8 @@
 
 namespace Micromus\KafkaBus\Consumers\Router;
 
-use Micromus\KafkaBus\Consumers\Messages\ConsumerMessage;
-use Micromus\KafkaBus\Interfaces\Messages\MessageFactoryInterface;
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageInterface;
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\MessageFactoryInterface;
 
 class Executor
 {
@@ -13,13 +13,13 @@ public function __construct(
     ) {
     }
 
-    public function execute(ConsumerMessage $message): void
+    public function execute(ConsumerMessageInterface $message): void
     {
         $this->handler
             ->execute($this->map($message));
     }
 
-    private function map(ConsumerMessage $message): mixed
+    private function map(ConsumerMessageInterface $message): mixed
     {
         return $this->factory
             ->fromKafka($message);

src/Consumers/Router/MessageFactory.php

@@ -2,7 +2,7 @@
 
 namespace Micromus\KafkaBus\Consumers\Router;
 
-use Micromus\KafkaBus\Messages\NativeMessageFactory;
+use Micromus\KafkaBus\Consumers\Messages\NativeMessageFactory;
 
 #[\Attribute]
 final readonly class MessageFactory

src/Consumers/Router/MessageFactoryClassExtractor.php

@@ -2,9 +2,9 @@
 
 namespace Micromus\KafkaBus\Consumers\Router;
 
-use Micromus\KafkaBus\Interfaces\Messages\MessageFactoryInterface;
+use Micromus\KafkaBus\Consumers\Messages\NativeMessageFactory;
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\MessageFactoryInterface;
 use Micromus\KafkaBus\Interfaces\ResolverInterface;
-use Micromus\KafkaBus\Messages\NativeMessageFactory;
 
 final class MessageFactoryClassExtractor
 {

src/Exceptions/Consumers/MessageConsumerNotHandledException.php

@@ -3,12 +3,12 @@
 namespace Micromus\KafkaBus\Exceptions\Consumers;
 
 use Exception;
-use Micromus\KafkaBus\Consumers\Messages\ConsumerMessage;
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageInterface;
 use Throwable;
 
 class MessageConsumerNotHandledException extends Exception
 {
-    public function __construct(public readonly ConsumerMessage $consumerMessage, ?Throwable $previous = null)
+    public function __construct(public readonly ConsumerMessageInterface $consumerMessage, ?Throwable $previous = null)
     {
         $message = "Message #{$this->consumerMessage->msgId()} from ".
             "{$this->consumerMessage->topicName()} not handled";

src/Interfaces/Bus/ThreadInterface.php

@@ -3,18 +3,18 @@
 namespace Micromus\KafkaBus\Interfaces\Bus;
 
 use Micromus\KafkaBus\Bus\Listeners\Listener;
-use Micromus\KafkaBus\Interfaces\Messages\MessageInterface;
 use Micromus\KafkaBus\Exceptions\Producers\RouteProducerException;
+use Micromus\KafkaBus\Interfaces\Producers\Messages\ProducerMessageInterface;
 
 interface ThreadInterface
 {
     /**
      * @throws RouteProducerException
      */
-    public function publish(MessageInterface $message): void;
+    public function publish(ProducerMessageInterface $message): void;
 
     /**
-     * @param  MessageInterface[]  $messages
+     * @param  ProducerMessageInterface[]  $messages
      *
      * @throws RouteProducerException
      */

src/Interfaces/Consumers/ConsumerInterface.php

@@ -2,8 +2,8 @@
 
 namespace Micromus\KafkaBus\Interfaces\Consumers;
 
-use Micromus\KafkaBus\Consumers\Messages\ConsumerMessage;
 use Micromus\KafkaBus\Exceptions\Consumers\ConsumerException;
+use Micromus\KafkaBus\Interfaces\Consumers\Messages\ConsumerMessageInterface;
 use Micromus\KafkaBus\Testing\Exceptions\KafkaMessagesEndedException;
 
 interface ConsumerInterface
@@ -12,7 +12,7 @@ interface ConsumerInterface
      * @throws KafkaMessagesEndedException
      * @throws ConsumerException
      */
-    public function getMessage(): ConsumerMessage;
+    public function getMessage(): ConsumerMessageInterface;
 
-    public function commit(ConsumerMessage $consumerMessage): void;
+    public function commit(ConsumerMessageInterface $consumerMessage): void;
 }