| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253 |
- 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"]
- KernelMode = Literal["auto", "20", "24"]
- @dataclass
- class RelayNode:
- name: str
- host: str
- port: int
- token: str
- weight: int = 100
- @dataclass
- class Config:
- relays: list[RelayNode]
- strategy: Strategy = "top3"
- kernel_mode: KernelMode = "auto"
- redundancy: int = 3
- tcp_warmup_bytes: int = 1048576
- probe_interval: float = 15.0
- tcp_loser_grace_ms: int = 1500
- direct_open_timeout: float = 10.0
- relay_open_timeout: float = 10.0
- tcp_connect_happy_eyeballs_delay: float | None = None
- relay_reconnect_delay: float = 3.0
- relay_tcp_nodelay: bool = True
- @classmethod
- def load(cls, path: str) -> "Config":
- raw = json.loads(Path(path).read_text())
- relays = [RelayNode(**item) for item in raw.get("relays", [])]
- return cls(
- relays=relays,
- strategy=raw.get("strategy", "top3"),
- kernel_mode=raw.get("kernel_mode", "auto"),
- redundancy=raw.get("redundancy", 3),
- tcp_warmup_bytes=raw.get("tcp_warmup_bytes", 1048576),
- probe_interval=raw.get("probe_interval", 15.0),
- tcp_loser_grace_ms=raw.get("tcp_loser_grace_ms", 1500),
- direct_open_timeout=raw.get("direct_open_timeout", 10.0),
- relay_open_timeout=raw.get("relay_open_timeout", 10.0),
- tcp_connect_happy_eyeballs_delay=raw.get("tcp_connect_happy_eyeballs_delay"),
- relay_reconnect_delay=raw.get("relay_reconnect_delay", 3.0),
- relay_tcp_nodelay=raw.get("relay_tcp_nodelay", True),
- )
|
