| 25 | self._is_initialized = False |
| 26 | |
| 27 | async def initialize(self) -> None: |
| 28 | if self._is_initialized: |
| 29 | return |
| 30 | |
| 31 | try: |
| 32 | self._conn = await aio_pika.connect_robust( |
| 33 | url=self.config.url, |
| 34 | heartbeat=self.config.heartbeat, |
| 35 | connection_timeout=self.config.connection_timeout, |
| 36 | ) |
| 37 | self._is_initialized = True |
| 38 | self.logger.info("🚀 RabbitMQ producer initialized successfully") |
| 39 | except Exception as e: |
| 40 | self.logger.error(f"🛑 Failed to initialize RabbitMQ producer: {e}") |
| 41 | raise |
| 42 | |
| 43 | async def close(self) -> None: |
| 44 | if self._conn and not self._conn.is_closed: |