fix: consumer middlewares work as expected [fixes #303] (#304)

Co-authored-by: Alexander (SASh) Alexiev <[email protected]>

src/Concerns/HandleConsumedMessage.php

@@ -15,7 +15,7 @@ trait HandleConsumedMessage
 {
     private function handleConsumedMessage(ConsumerMessage $message, Handler|Closure $handler, ?MessageConsumer $consumer = null, array $middlewares = []): void
     {
-        $middlewares = array_map($this->wrapMiddleware(...), $middlewares);
+        $middlewares = array_map(fn ($middleware) => $this->wrapMiddleware($middleware, $consumer), $middlewares);
         $middlewares = array_reverse($middlewares);
 
         foreach ($middlewares as $middleware) {

src/Concerns/PrepareMiddlewares.php

@@ -2,13 +2,14 @@
 
 namespace Junges\Kafka\Concerns;
 
+use Junges\Kafka\Contracts\MessageConsumer;
 use Junges\Kafka\Contracts\Middleware;
 
 /** @internal */
 trait PrepareMiddlewares
 {
     /** Wrap the message with a given middleware. */
-    private function wrapMiddleware(Middleware|string|callable $middleware): callable
+    private function wrapMiddleware(Middleware|string|callable $middleware, ?MessageConsumer $consumer = null): callable
     {
         $middleware = match(true) {
             is_string($middleware) && is_subclass_of($middleware, Middleware::class) => new $middleware(),
@@ -17,6 +18,6 @@ private function wrapMiddleware(Middleware|string|callable $middleware): callabl
             default => throw new \LogicException('Invalid middleware.')
         };
 
-        return static fn (callable $handler) => static fn ($message) => $middleware($message, $handler);
+        return static fn (callable $handler) => static fn ($message) => $middleware($message, fn($message) => $handler($message, $consumer));
     }
 }