| 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 |