import socket
from dataclasses import dataclass
from logging import Logger
from ..infra import depends_on
from ..config.full_node import FullNodeConfig
from .config_provider import ConfigProvider
[docs]
@dataclass
class PortManager:
"""Changes port if already in use by another process."""
log: Logger
config: ConfigProvider | FullNodeConfig
[docs]
@depends_on("config")
def start(self):
self.acquire_port()
[docs]
def acquire_port(self):
base_url_is_derived = (self.config.koi_net.node_profile.base_url == self.config.server.url)
changed_port: bool = False
while True:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
if s.connect_ex((self.config.server.host, self.config.server.port)) != 0:
break
self.log.debug(f"Port {self.config.server.port} in use")
self.config.server.port += 1
changed_port = True
self.log.debug(f"Acquired port {self.config.server.port}")
if base_url_is_derived and changed_port:
self.log.debug("Updating node profile")
self.config.koi_net.node_profile.base_url = self.config.server.url
self.config.save_to_yaml()