Repository: Significant-Gravitas/Auto-GPT
File: autogpt_platform/backend/backend/data/rabbitmq.py (Python)

Code before:
    parameters = pika.ConnectionParameters(
        host=HOST,
        port=PORT,
        virtual_host=VIRTUAL_HOST,
        credentials=credentials,
        heartbeat=600,
        blocked_connection_timeout=300,
    )

    connection = pika.BlockingConnection(parameters)
    channel = connection.channel()
    channel.basic_qos(prefetch_count=1)
    return connection


@conn_retry("RabbitMQ", "Releasing connection")
def disconnect():
    global connection, channel
    if channel:
        if channel.is_open:
            channel.close()
        channel = None
    if connection:
        if connection.is_open:
            connection.close()
        connection = None


def get_channel(auto_connect: bool = True) -> BlockingChannel:
    global channel
    if channel and channel.is_open:
        return channel
    if auto_connect:
        connect()
        # This may seem like a wierd place to double check, but its a good sanity check
        # and fixes the type checker since we are relying on globals
        if not channel:
            raise RuntimeError("RabbitMQ connection is not established")
        return channel
    raise RuntimeError("RabbitMQ connection is not established")


def declare_queue(
    queue_name: str, durable: bool = True, auto_delete: bool = False
) -> None:
    ch = get_channel()
    ch.queue_declare(queue=queue_name, durable=durable, auto_delete=auto_delete)


def publish_message(
    queue_name: str,

Review comment (suggestion):
```
if channel and channel.is_open:
    channel.close()
channel = None
```
behaves the same and remove the nested code.

Code after:
    vhost: str = "/"
    exchanges: list[Exchange]
    queues: list[Queue]


class RabbitMQBase(ABC):
    """Base class for RabbitMQ connections with shared configuration"""

    def __init__(self, config: RabbitMQConfig):
        settings = Settings()
        self.host = settings.config.rabbitmq_host
        self.port = settings.config.rabbitmq_port
        self.username = settings.secrets.rabbitmq_default_user
        self.password = settings.secrets.rabbitmq_default_pass
        self.config = config

        self._connection = None
        self._channel = None

    @property
    def is_connected(self) -> bool:
        """Check if we have a valid connection"""
        return bool(self._connection)

    @property
    def is_ready(self) -> bool:
        """Check if we have a valid channel"""
        return bool(self.is_connected and self._channel)

    @abstractmethod
    def connect(self) -> None | Awaitable[None]:
        """Establish connection to RabbitMQ"""
        pass

    @abstractmethod
    def disconnect(self) -> None | Awaitable[None]:
        """Close connection to RabbitMQ"""
        pass

    @abstractmethod
    def declare_infrastructure(self) -> None | Awaitable[None]:
        """Declare exchanges and queues for this service"""
        pass


class SyncRabbitMQ(RabbitMQBase):
    """Synchronous RabbitMQ client"""

    @property
    def is_connected(self) -> bool:
        return bool(self._connection and self._connection.is_open)