| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475 |
- from __future__ import annotations
- import json
- from dataclasses import dataclass
- from pathlib import Path
- from typing import Literal
- Strategy = Literal["broadcast", "top2", "top3", "top4", "backup"]
- @dataclass
- class UdpRelayNode:
- name: str
- host: str
- port: int
- token: str
- weight: int = 100
- @dataclass
- class UdpConfig:
- relays: list[UdpRelayNode]
- strategy: Strategy = "top3"
- redundancy: int = 3
- probe_interval: float = 15.0
- relay_open_timeout: float = 10.0
- relay_reconnect_delay: float = 3.0
- relay_reconnect_attempts: int = 5
- relay_reconnect_max_delay: float = 30.0
- relay_ping_interval: float = 10.0
- relay_ping_timeout: float = 25.0
- relay_tcp_nodelay: bool = True
- udp_redundancy: int = 2
- udp_direct_copies: int | None = None
- udp_relay_copies: int | None = None
- udp_direct_redundancy: int = 3
- udp_direct_redundancy_v4: int | None = None
- udp_direct_redundancy_v6: int | None = None
- udp_always_broadcast: bool = True
- udp_copy_interval_ms: int = 2
- udp_failover_idle_ms: int = 700
- socks_host: str = "127.0.0.1"
- socks_port: int = 0
- @classmethod
- def from_dict(cls, raw: dict) -> "UdpConfig":
- relays = [UdpRelayNode(**item) for item in raw.get("relays", [])]
- return cls(
- relays=relays,
- strategy=raw.get("strategy", "top3"),
- redundancy=raw.get("redundancy", 3),
- probe_interval=raw.get("probe_interval", 15.0),
- relay_open_timeout=raw.get("relay_open_timeout", 10.0),
- relay_reconnect_delay=raw.get("relay_reconnect_delay", 3.0),
- relay_reconnect_attempts=max(1, raw.get("relay_reconnect_attempts", 5)),
- relay_reconnect_max_delay=max(raw.get("relay_reconnect_delay", 3.0), raw.get("relay_reconnect_max_delay", 30.0)),
- relay_ping_interval=max(1.0, raw.get("relay_ping_interval", 10.0)),
- relay_ping_timeout=max(1.0, raw.get("relay_ping_timeout", 25.0)),
- relay_tcp_nodelay=raw.get("relay_tcp_nodelay", True),
- udp_redundancy=max(0, raw.get("udp_redundancy", 2)),
- udp_direct_copies=max(1, raw["udp_direct_copies"]) if raw.get("udp_direct_copies") is not None else None,
- udp_relay_copies=max(1, raw["udp_relay_copies"]) if raw.get("udp_relay_copies") is not None else None,
- udp_direct_redundancy=max(1, raw.get("udp_direct_redundancy", 3)),
- udp_direct_redundancy_v4=raw.get("udp_direct_redundancy_v4"),
- udp_direct_redundancy_v6=raw.get("udp_direct_redundancy_v6"),
- udp_always_broadcast=raw.get("udp_always_broadcast", True),
- udp_copy_interval_ms=max(0, raw.get("udp_copy_interval_ms", 2)),
- udp_failover_idle_ms=max(100, raw.get("udp_failover_idle_ms", 700)),
- socks_host=raw.get("socks_host", "127.0.0.1"),
- socks_port=max(0, raw.get("socks_port", 0)),
- )
- @classmethod
- def load(cls, path: str) -> "UdpConfig":
- return cls.from_dict(json.loads(Path(path).read_text()))
|
