Source code for edri.switch.connection

from dataclasses import dataclass, field
from threading import Event

from edri.protocol.channel import SecureChannel
from edri.switch.mailbox import Mailbox




@dataclass
class Connection:
    mailbox: Mailbox
    channel: SecureChannel | None
    demands: bytes | None = None
    key_index: int = 0
    stop: Event = field(default_factory=Event)

    @property
    def queue(self) -> Mailbox:
        """Compatibility alias for the old connection queue."""
        return self.mailbox



    def disconnect(self) -> None:
        self.stop.set()
        self.mailbox.wake()
        if self.channel is not None:
            self.channel.close()