using System.Buffers.Text; using System.Collections.Immutable; using System.Text; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Proxd_man.Config; using Proxd_man.Config.Outbound; using Proxd_man.ShareLink; namespace Proxd_man.Subscribe; public class SubscribeManager { private readonly Dictionary> _subscribeOutbounds = new(); private CancellationTokenSource? _cts; private readonly ILogger _logger; private readonly IOptionsMonitor _config; public SubscribeManager(ILogger logger, IOptionsMonitor config) { _logger = logger; _config = config; _config.OnChange((_, _) => Task.Run(async () => await CheckAndUpdateSubscribes())); } public delegate void SubscribeUpdatedEventHandler(IDictionary> subscribeOutbounds); public event SubscribeUpdatedEventHandler? SubscribeUpdated; private ProxdConfig Config => _config.Get(null); public async Task CheckAndUpdateSubscribes(CancellationToken cancellationToken = default) { var httpClient = new HttpClient(); _logger.LogInformation("开始更新订阅..."); var newSubscribeOutbounds = new Dictionary>(); for (var i = 0; i < Config.Subscribe.Subscribes.Count; i++) { var subscribe = Config.Subscribe.Subscribes[i]; var outbounds = new List(); _logger.LogInformation("正在更新订阅 {subscribeName} ({current}/{total})...", subscribe.Name, i + 1, Config.Subscribe.Subscribes.Count); try { var response = await httpClient.GetAsync(subscribe.Url, cancellationToken); response.EnsureSuccessStatusCode(); var content = await response.Content.ReadAsStringAsync(cancellationToken); if (Base64.IsValid(content)) { content = Encoding.UTF8.GetString(Convert.FromBase64String(content)); } var shareLinks = content.Split('\n').Select(l => l.Trim()).Where(l => !string.IsNullOrEmpty(l)); foreach (var shareLink in shareLinks) { try { var outbound = ShareLinkParser.Parse(shareLink); outbounds.Add(outbound); } catch (InvalidShareLinkException e) { _logger.LogWarning(e, "解析分享链接失败: {shareLink}\n", shareLink); } } } catch (Exception e) { _logger.LogWarning(e, "更新订阅失败: {subscribeName}", subscribe.Name); } newSubscribeOutbounds[subscribe] = outbounds; _logger.LogInformation("订阅 {subscribeName} 更新完成. (共 {outboundCount} 个节点)", subscribe.Name, outbounds.Count); } _subscribeOutbounds.Clear(); foreach (var (source, outbounds) in newSubscribeOutbounds) { _subscribeOutbounds[source] = outbounds; } _logger.LogInformation("订阅更新完成."); SubscribeUpdated?.Invoke(_subscribeOutbounds.ToImmutableDictionary()); } public IDictionary> GetSubscribeOutbounds() { return _subscribeOutbounds.ToImmutableDictionary(); } public void StartUpdateSubscribes() { var cts = new CancellationTokenSource(); Task.Run(async () => { while (!cts.IsCancellationRequested) { await CheckAndUpdateSubscribes(cts.Token); await Task.Delay(TimeSpan.FromHours(12), cts.Token); } }, cts.Token); _cts = cts; _logger.LogInformation("订阅更新任务已启动."); } public void StopUpdateSubscribes() { _cts?.Cancel(); } }