using System.Diagnostics; using System.Net.Sockets; using System.Text; using System.Text.Json; using Microsoft.Extensions.Options; namespace MiGu.Server.Ota; public sealed class WatchDogClient { private readonly IHttpClientFactory _httpFactory; private readonly OtaOptions _opt; private readonly ILogger _log; private readonly JsonSerializerOptions _json = new() { PropertyNameCaseInsensitive = true }; public WatchDogClient(IHttpClientFactory httpFactory, IOptions opt, ILogger log) { _httpFactory = httpFactory; _opt = opt.Value; _log = log; } private HttpClient CreateClient(int? timeoutMs = null) { var c = _httpFactory.CreateClient(nameof(WatchDogClient)); c.Timeout = TimeSpan.FromMilliseconds(timeoutMs ?? _opt.RequestTimeoutMs); return c; } private string Base(string ip) => $"http://{ip}:{_opt.WatchDogPort}"; public async Task<(bool Ok, OtaAppVersions? M, OtaAppVersions? D, OtaAppVersions? C, string? Error)> GetMdcInfoAsync(string ip, CancellationToken ct) { try { using var client = CreateClient(); using var resp = await client.GetAsync($"{Base(ip)}/getMDCInfo", ct); if (!resp.IsSuccessStatusCode) return (false, null, null, null, $"HTTP {(int)resp.StatusCode}"); var text = await resp.Content.ReadAsStringAsync(ct); using var doc = JsonDocument.Parse(text); var root = doc.RootElement; return (true, ParseApp(root, "Medulla"), ParseApp(root, "Detour"), ParseApp(root, "Clumsy"), null); } catch (Exception ex) { return (false, null, null, null, ex.Message); } } private static OtaAppVersions? ParseApp(JsonElement root, string name) { if (!root.TryGetProperty(name, out var app) && !root.TryGetProperty(name.ToLowerInvariant(), out app)) return null; return new OtaAppVersions { Exe = ParseComp(app, "exe"), Dll = ParseComp(app, "dll"), Pdb = ParseComp(app, "pdb") }; } private static OtaComponentVersion? ParseComp(JsonElement app, string key) { if (!app.TryGetProperty(key, out var c)) return null; string? ver = null; string? time = null; if (c.ValueKind == JsonValueKind.Object) { if (c.TryGetProperty("version", out var v)) ver = v.ValueKind == JsonValueKind.String ? v.GetString() : v.ToString(); if (c.TryGetProperty("time", out var t)) time = t.GetString(); } else if (c.ValueKind == JsonValueKind.String) { ver = c.GetString(); } return new OtaComponentVersion { Version = ver, Time = time }; } public async Task MeasureRttMsAsync(string ip, CancellationToken ct) { try { var sw = Stopwatch.StartNew(); using var tcp = new TcpClient(); using var reg = ct.Register(() => { try { tcp.Close(); } catch { /* ignore */ } }); var connectTask = tcp.ConnectAsync(ip, _opt.WatchDogPort); var done = await Task.WhenAny(connectTask, Task.Delay(Math.Min(3000, _opt.RequestTimeoutMs), ct)); if (done != connectTask || !tcp.Connected) return null; await connectTask; sw.Stop(); return (int)sw.ElapsedMilliseconds; } catch { return null; } } public async Task TriggerPullAsync(string ip, string serverBaseUrl, string time, CancellationToken ct) { // 现网 WatchDog 忽略 server 查询参数,固定 POST 到 http://{config.serverIP}:8000/upload-mdcs/{key}。 // serverBaseUrl 仅作日志/未来兼容;真正要通必须:车上 serverIP=本机局域网 IP,且本机监听 ReceivePort。 using var client = CreateClient(_opt.UploadTimeoutMs); var url = $"{Base(ip)}/getmdcsexe?time={Uri.EscapeDataString(time)}&server={Uri.EscapeDataString(serverBaseUrl.TrimEnd('/'))}"; try { using var resp = await client.GetAsync(url, ct); var body = (await resp.Content.ReadAsStringAsync(ct)).Trim(); _log.LogInformation("getmdcsexe {Ip} -> {Code} body={Body} (WatchDog will POST to its config.serverIP:{Port}/upload-mdcs/*; expect receiver {Base})", ip, (int)resp.StatusCode, body.Length > 200 ? body[..200] : body, _opt.ReceivePort, serverBaseUrl); if (!resp.IsSuccessStatusCode) throw new InvalidOperationException($"WatchDog getmdcsexe HTTP {(int)resp.StatusCode}: {body}"); if (body.Contains("请配置", StringComparison.Ordinal) || body.Equals("false", StringComparison.OrdinalIgnoreCase) || body.Equals("\"false\"", StringComparison.OrdinalIgnoreCase)) { throw new InvalidOperationException( "WatchDog 拒绝拉包或回传失败。请确认:1) 车已配置 Medulla/Detour/Clumsy 路径;" + $"2) watch_dog.json 的 serverIP 指向本机局域网 IP(车将 POST 到 serverIP:{_opt.ReceivePort}/upload-mdcs/*);" + $"3) 本机已监听 :{_opt.ReceivePort}。WatchDog 返回:{body}"); } } catch (InvalidOperationException) { throw; } catch (Exception ex) { _log.LogWarning(ex, "getmdcsexe failed {Ip}", ip); throw; } } public async Task UploadComponentAsync(string ip, string componentKey, string localPath, string fileName, int bandwidthKbps, CancellationToken ct) { var endpoint = OtaPathMap.WatchDogUpdatePath(componentKey) ?? throw new ArgumentException($"未知组件 {componentKey}"); await UploadFileAsync($"{Base(ip)}/{endpoint}", localPath, fileName, bandwidthKbps, null, ct); } public async Task UploadCustomFileAsync(string ip, string localPath, string fileName, string remotePath, int restartOp, int bandwidthKbps, CancellationToken ct) { var url = $"{Base(ip)}/updateFile/{Uri.EscapeDataString(fileName)}/{restartOp}/"; await UploadFileAsync(url, localPath, fileName, bandwidthKbps, new Dictionary { ["path"] = remotePath }, ct); } private async Task UploadFileAsync(string url, string localPath, string fileName, int bandwidthKbps, Dictionary? extraFields, CancellationToken ct) { using var client = CreateClient(_opt.UploadTimeoutMs); await using var fs = File.OpenRead(localPath); Stream contentStream = fs; if (bandwidthKbps > 0) contentStream = new ThrottledStream(fs, bandwidthKbps * 1024L); using var form = new MultipartFormDataContent(); if (extraFields != null) { foreach (var (k, v) in extraFields) form.Add(new StringContent(v, Encoding.UTF8), k); } var streamContent = new StreamContent(contentStream); form.Add(streamContent, "file", fileName); using var resp = await client.PostAsync(url, form, ct); if (!resp.IsSuccessStatusCode) { var body = await resp.Content.ReadAsStringAsync(ct); throw new InvalidOperationException($"上传失败 HTTP {(int)resp.StatusCode}: {body}"); } } public async Task GetJsonAsync(string ip, string app, CancellationToken ct) { var path = app.ToLowerInvariant() switch { "medulla" => "getMedullajson", "detour" => "getDetourjson", "clumsy" => "getClumsyjson", _ => throw new ArgumentException("app 须为 medulla|detour|clumsy") }; using var client = CreateClient(); using var resp = await client.GetAsync($"{Base(ip)}/{path}", ct); resp.EnsureSuccessStatusCode(); return await resp.Content.ReadAsStringAsync(ct); } public async Task PutJsonAsync(string ip, string app, string json, CancellationToken ct) { var path = app.ToLowerInvariant() switch { "medulla" => "updateMedullajson", "detour" => "updateDetourjson", "clumsy" => "updateClumsyjson", _ => throw new ArgumentException("app 须为 medulla|detour|clumsy") }; using var client = CreateClient(); using var content = new StringContent(json, Encoding.UTF8, "application/json"); using var resp = await client.PostAsync($"{Base(ip)}/{path}", content, ct); if (!resp.IsSuccessStatusCode) { var body = await resp.Content.ReadAsStringAsync(ct); throw new InvalidOperationException($"更新 JSON 失败 HTTP {(int)resp.StatusCode}: {body}"); } } /// 简易限速流:按字节/秒节流读取。 private sealed class ThrottledStream : Stream { private readonly Stream _inner; private readonly long _bytesPerSecond; private long _windowBytes; private long _windowStart = Environment.TickCount64; public ThrottledStream(Stream inner, long bytesPerSecond) { _inner = inner; _bytesPerSecond = Math.Max(1024, bytesPerSecond); } public override bool CanRead => _inner.CanRead; public override bool CanSeek => false; public override bool CanWrite => false; public override long Length => _inner.Length; public override long Position { get => _inner.Position; set => throw new NotSupportedException(); } public override void Flush() => _inner.Flush(); public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); public override void SetLength(long value) => throw new NotSupportedException(); public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); public override int Read(byte[] buffer, int offset, int count) { var n = _inner.Read(buffer, offset, count); if (n > 0) Throttle(n); return n; } public override async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) { var n = await _inner.ReadAsync(buffer.AsMemory(offset, count), cancellationToken); if (n > 0) await ThrottleAsync(n, cancellationToken); return n; } private void Throttle(int n) { _windowBytes += n; var elapsed = Environment.TickCount64 - _windowStart; if (elapsed < 1) elapsed = 1; var allowed = _bytesPerSecond * elapsed / 1000; if (_windowBytes > allowed) { var wait = (int)((_windowBytes - allowed) * 1000 / _bytesPerSecond); if (wait > 0) Thread.Sleep(Math.Min(wait, 2000)); } if (elapsed >= 1000) { _windowBytes = 0; _windowStart = Environment.TickCount64; } } private async Task ThrottleAsync(int n, CancellationToken ct) { _windowBytes += n; var elapsed = Environment.TickCount64 - _windowStart; if (elapsed < 1) elapsed = 1; var allowed = _bytesPerSecond * elapsed / 1000; if (_windowBytes > allowed) { var wait = (int)((_windowBytes - allowed) * 1000 / _bytesPerSecond); if (wait > 0) await Task.Delay(Math.Min(wait, 2000), ct); } if (elapsed >= 1000) { _windowBytes = 0; _windowStart = Environment.TickCount64; } } protected override void Dispose(bool disposing) { // 不释放 inner(由调用方 using FileStream) base.Dispose(disposing); } } }