| 15 | |
| 16 | |
| 17 | class AsyncRabbitMQConsumer: |
| 18 | def __init__(self, config: RabbitMQConfig, log: Log) -> None: |
| 19 | self.config = config |
| 20 | self.logger = log |
| 21 | self._conn: AbstractConnection | None = None |
| 22 | self._channel: dict[str, AbstractChannel] = {} |
| 23 | self._handlers: dict[str, MessageHandler] = {} |
| 24 | self._is_initialized = False |
| 25 | self._running = False |
| 26 | self._shutdown_event = asyncio.Event() |
| 27 | |
| 28 | async def initialize( |
| 29 | self, |
| 30 | loop: asyncio.AbstractEventLoop, |
| 31 | ) -> None: |
| 32 | if self._is_initialized: |
| 33 | return |
| 34 | |
| 35 | try: |
| 36 | self._conn = await aio_pika.connect_robust( |
| 37 | url=self.config.url, |
| 38 | heartbeat=self.config.heartbeat, |
| 39 | connection_timeout=self.config.connection_timeout, |
| 40 | loop=loop, |
| 41 | ) |
| 42 | self._is_initialized = True |
| 43 | self.logger.info("🚀 RabbitMQ consumer initialized successfully") |
| 44 | except Exception as e: |
| 45 | self.logger.error(f"🛑 Failed to initialize RabbitMQ consumer: {e}") |
| 46 | raise |
| 47 | |
| 48 | async def close(self) -> None: |
| 49 | self._running = False |
| 50 | self._shutdown_event.set() |
| 51 | |
| 52 | for channel in self._channel.values(): |
| 53 | if not channel.is_closed: |
| 54 | await channel.close() |
| 55 | |
| 56 | if self._conn and not self._conn.is_closed: |
| 57 | await self._conn.close() |
| 58 | |
| 59 | self._is_initialized = False |
| 60 | self._channel.clear() |
| 61 | self.logger.info("🚦 RabbitMQ consumer closed") |
| 62 | |
| 63 | def register_handler(self, handler: MessageHandler) -> None: |
| 64 | handler_name = handler.__class__.__name__ |
| 65 | self._handlers[handler_name] = handler |
| 66 | self.logger.info(f"📨 RabbitMQ consumer registered handler: {handler_name}") |
| 67 | |
| 68 | def get_handler(self, action: str) -> MessageHandler | None: |
| 69 | for handler in self._handlers.values(): |
| 70 | if handler.can_handle(action): |
| 71 | return handler |
| 72 | return None |
| 73 | |
| 74 | async def _create_channel(self, queue_name: str) -> AbstractChannel: |
nothing calls this directly
no outgoing calls
no test coverage detected