Method consume

src/core/rabbit_mq/consumer.py:140–182

(
        self,
        queue_name: str,
        exchange_name: str,
        durable: bool = True,
        exclusive: bool = False,
        auto_delete: bool = False,
    )

            await message.reject(requeue=False)

    async def consume(
        self,
        queue_name: str,
        exchange_name: str,
        durable: bool = True,
        exclusive: bool = False,
        auto_delete: bool = False,
    ) -> None:
        if not self._is_initialized or not self._conn:
            raise RuntimeError("🛑 RabbitMQ consumer is not initialized")
        try:
            async with self._conn:
                channel = await self._conn.channel()  # type: ignore
                await channel.set_qos(prefetch_count=1)

                exchange = await channel.declare_exchange(
                    name=exchange_name,
                    type=ExchangeType.DIRECT,
                    durable=durable,
                    auto_delete=auto_delete,
                    internal=False,
                )

                queue = await channel.declare_queue(
                    queue_name,
                    durable=durable,
                    exclusive=exclusive,
                    auto_delete=auto_delete,
                )

                await queue.bind(exchange, queue_name)

                await queue.consume(lambda message: self._process_message(message, queue_name), no_ack=False)

                self._running = True
                self.logger.info(f"🚀 RabbitMQ consumer started consuming messages from queue: {queue_name}")
                await self._shutdown_event.wait()
        except Exception as e:
            self.logger.error(
                f"🛑 Failed to consume messages from queue {queue_name}: {e}",
                error=traceback.extract_tb(e.__traceback__)[-1],
            )
            raise