Fixed formats, added log.

Signed-off-by: Pavel Kirilin <[email protected]>

Author: s3rius

taskiq_aio_pika/taskiq/brokers/aio_pika_broker.py

@@ -1,4 +1,5 @@
 from asyncio import AbstractEventLoop
+from logging import getLogger
 from typing import Any, AsyncGenerator, Optional, TypeVar
 
 from aio_pika import Channel, ExchangeType, Message, connect_robust
@@ -10,6 +11,8 @@
 
 _T = TypeVar("_T")
 
+logger = getLogger("taskiq.aio_pika_broker")
+
 
 class AioPikaBroker(AsyncBroker):
     def __init__(
@@ -84,12 +87,18 @@ async def listen(self) -> AsyncGenerator[BrokerMessage, None]:
                 async for rmq_message in queue_iter:
                     async with rmq_message.process():
                         try:
-                            yield BrokerMessage.parse_raw(
-                                rmq_message.body,
-                                content_type=rmq_message.content_type or "",
+                            yield BrokerMessage(
+                                task_id=rmq_message.headers["task_id"],
+                                task_name=rmq_message.headers["task_name"],
+                                message=rmq_message.body,
+                                headers=rmq_message.headers,
+                            )
+                        except (ValueError, LookupError) as exc:
+                            logger.debug(
+                                "Cannot read broker message %s",
+                                exc,
+                                exc_info=True,
                             )
-                        except ValueError:
-                            continue
 
     async def shutdown(self) -> None:
         await self.connection_pool.close()