diff options
Diffstat (limited to 'stacks/daemon/modular_tcp_server.py')
| -rw-r--r-- | stacks/daemon/modular_tcp_server.py | 208 |
1 files changed, 208 insertions, 0 deletions
diff --git a/stacks/daemon/modular_tcp_server.py b/stacks/daemon/modular_tcp_server.py new file mode 100644 index 0000000..b7f5612 --- /dev/null +++ b/stacks/daemon/modular_tcp_server.py @@ -0,0 +1,208 @@ +""" +Modular TCP/IP Servers Service — central Main + Secondary routing for MAX25. + +Public name in docs: "modular TCP/IP Servers Service". +Main instance accepts M25/1 clients and forwards sessions to Secondary max25d peers. +""" +from __future__ import annotations + +import configparser +import selectors +import socket +import threading +from dataclasses import dataclass, field +from typing import Callable, Optional + + +LogFn = Callable[[str], None] + + +@dataclass +class SecondaryEndpoint: + name: str + host: str + port: int + + +@dataclass +class ModularTcpConfig: + enabled: bool = False + role: str = "standalone" # standalone | main | secondary + service_name: str = "modular-tcp-server" + instance_id: str = "" + secondaries: list[SecondaryEndpoint] = field(default_factory=list) + connect_timeout: float = 5.0 + buffer_size: int = 65536 + + +def load_modular_tcp(cp: configparser.ConfigParser) -> ModularTcpConfig: + cfg = ModularTcpConfig() + if not cp.has_section("modular_tcp"): + return cfg + sec = cp["modular_tcp"] + cfg.enabled = sec.get("enabled", "no").strip().lower() in ("1", "yes", "true", "on") + cfg.role = sec.get("role", cfg.role).strip().lower() or cfg.role + cfg.service_name = sec.get("service_name", cfg.service_name).strip() or cfg.service_name + cfg.instance_id = sec.get("instance_id", cfg.instance_id).strip() + if cp.has_option("modular_tcp", "connect_timeout"): + try: + cfg.connect_timeout = float(cp.get("modular_tcp", "connect_timeout")) + except (TypeError, ValueError): + cfg.connect_timeout = 5.0 + if cfg.connect_timeout <= 0.0: + cfg.connect_timeout = 5.0 + if cp.has_section("modular_tcp.secondaries"): + for key in cp.options("modular_tcp.secondaries"): + raw = cp.get("modular_tcp.secondaries", key).strip() + if not raw: + continue + host, _, port_s = raw.partition(":") + host = host.strip() + if not host: + continue + try: + port = int(port_s or "7326") + except (TypeError, ValueError): + port = 7326 + if port < 1 or port > 65535: + port = 7326 + cfg.secondaries.append(SecondaryEndpoint(name=key.strip(), host=host, port=port)) + if cfg.enabled and cfg.role == "main" and not cfg.secondaries: + # Legacy comma list: secondaries = host:port,host:port + raw = sec.get("secondaries", "").strip() + if raw: + for idx, item in enumerate(x.strip() for x in raw.split(",") if x.strip()): + host, _, port_s = item.partition(":") + host = host.strip() + if not host: + continue + try: + port = int(port_s or "7326") + except (TypeError, ValueError): + port = 7326 + if port < 1 or port > 65535: + port = 7326 + cfg.secondaries.append( + SecondaryEndpoint( + name=f"secondary-{idx + 1}", + host=host, + port=port, + ) + ) + return cfg + + +class ModularTcpMainService: + """Bidirectional relay: operator client <-> Secondary max25d M25/1.""" + + def __init__( + self, + cfg: ModularTcpConfig, + listen_host: str, + listen_port: int, + log: LogFn, + ) -> None: + self.cfg = cfg + self.listen_host = listen_host + self.listen_port = listen_port + self.log = log + self._stop = threading.Event() + self._thread: Optional[threading.Thread] = None + self._rr_index = 0 + self._lock = threading.Lock() + + def _pick_secondary(self) -> Optional[SecondaryEndpoint]: + with self._lock: + if not self.cfg.secondaries: + return None + ep = self.cfg.secondaries[self._rr_index % len(self.cfg.secondaries)] + self._rr_index += 1 + return ep + + def _relay(self, client: socket.socket, upstream: socket.socket) -> None: + sel = selectors.DefaultSelector() + sel.register(client, selectors.EVENT_READ) + sel.register(upstream, selectors.EVENT_READ) + try: + while not self._stop.is_set(): + for key, _ in sel.select(timeout=0.5): + sock = key.fileobj + assert isinstance(sock, socket.socket) + try: + data = sock.recv(self.cfg.buffer_size) + except OSError: + return + if not data: + return + target = upstream if sock is client else client + try: + target.sendall(data) + except OSError: + return + finally: + sel.close() + + def _handle_client(self, client: socket.socket, addr: tuple[str, int]) -> None: + ep = self._pick_secondary() + if ep is None: + self.log(f"modular_tcp: no secondaries configured — drop {addr}") + client.close() + return + upstream = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + upstream.settimeout(self.cfg.connect_timeout) + try: + upstream.connect((ep.host, ep.port)) + except OSError as exc: + self.log(f"modular_tcp: upstream {ep.name} {ep.host}:{ep.port} failed ({exc})") + client.close() + return + upstream.settimeout(None) + self.log(f"modular_tcp: {addr} -> {ep.name} {ep.host}:{ep.port}") + try: + self._relay(client, upstream) + finally: + for s in (client, upstream): + try: + s.shutdown(socket.SHUT_RDWR) + except OSError: + pass + s.close() + + def _serve(self) -> None: + srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + srv.bind((self.listen_host, self.listen_port)) + srv.listen(64) + srv.settimeout(1.0) + self.log( + f"modular_tcp: Main service '{self.cfg.service_name}' " + f"listen {self.listen_host}:{self.listen_port} " + f"secondaries={len(self.cfg.secondaries)}" + ) + try: + while not self._stop.is_set(): + try: + client, addr = srv.accept() + except socket.timeout: + continue + except OSError: + break + threading.Thread( + target=self._handle_client, + args=(client, addr), + daemon=True, + ).start() + finally: + srv.close() + + def start(self) -> None: + if self._thread is not None: + return + self._thread = threading.Thread(target=self._serve, name="modular-tcp-main", daemon=True) + self._thread.start() + + def stop(self) -> None: + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=3.0) + self._thread = None |