using AllBase; using LaunchDarkly.EventSource; using Newtonsoft.Json; using OrderSupportTicket.Models; using OrderSupportTicket.Models.Api; using System; using System.Diagnostics; using System.Linq; using System.Net; using System.Net.Http; using System.Threading.Tasks; using WxOrder.Framework.Helpers; namespace OrderSupportTicket.Services; /// /// 基于 LaunchDarkly.EventSource 的 SSE 客户端。 /// net462 可使用;断线自动退避重连,并携带服务端事件 ID 进行补偿。 /// public sealed class EventSourceTicketRealtimeService { private const string EventsPath = "/api/open/v2/customer-work-orders/events"; private readonly object _sync = new object(); private EventSource _eventSource; private bool _running; private bool _disposed; public event EventHandler NotificationReceived; public event EventHandler StateChanged; /// /// 当前 SSE 连接状态。 /// public TicketRealtimeState State { get; private set; } = TicketRealtimeState.Disconnected; public Task Start(TicketClientOptions options) { if (_disposed) throw new ObjectDisposedException(nameof(EventSourceTicketRealtimeService)); if (options == null) throw new ArgumentNullException(nameof(options)); TicketApiService.ValidateOptions(options); StopCore(); var config = BuildEventsUri(options); var source = new EventSource(config); source.Opened += OnOpened; source.Closed += OnClosed; source.MessageReceived += OnMessageReceived; source.Error += OnError; lock (_sync) { _running = true; _eventSource = source; } SetState(TicketRealtimeState.Connecting, "连接中…"); _= RunAsync(source); return Task.FromResult(0); } public Task StopAsync() { StopCore(); SetState(TicketRealtimeState.Disconnected, "已断开"); return Task.FromResult(0); } public void Dispose() { if (_disposed) return; StopCore(); _disposed = true; } private async Task RunAsync(EventSource source) { try { await source.StartAsync(); lock (_sync) { if (!_running || !ReferenceEquals(_eventSource, source)) return; } SetState(TicketRealtimeState.Reconnecting, "连接已关闭,等待重连…"); } catch (Exception ex) { NLogger.Instance.Error("工单服务连接失败:" + ex.Message); lock (_sync) { if (!_running || !ReferenceEquals(_eventSource, source)) return; } SetState(TicketRealtimeState.Reconnecting, "连接失败:" + ex.Message); } } private void OnOpened(object sender, StateChangedEventArgs args) { SetState(TicketRealtimeState.Connected, "已连接"); } private void OnClosed(object sender, StateChangedEventArgs args) { lock (_sync) { if (!_running || !ReferenceEquals(_eventSource, sender)) return; } SetState(TicketRealtimeState.Reconnecting, "连接断开,重连中…"); } private void OnError(object sender, ExceptionEventArgs args) { lock (_sync) { if (!_running || !ReferenceEquals(_eventSource, sender)) return; } SetState( TicketRealtimeState.Reconnecting, "连接异常,重连中:" + (args.Exception?.Message ?? "未知错误")); } private void OnMessageReceived(object sender, MessageReceivedEventArgs args) { if (args == null) return; var data = args.Message.Data; if (string.IsNullOrWhiteSpace(data)) return; TicketRealtimeNotification notification; try { Debug.WriteLine(data); notification = JsonConvert.DeserializeObject(data, TicketApiService.JsonSettings); } catch (JsonException) { return; } if (notification == null) return; if (string.IsNullOrWhiteSpace(notification.EventId)) notification.EventId = args.Message.LastEventId; if (string.IsNullOrWhiteSpace(notification.Type)) notification.Type = args.EventName; NotificationReceived?.Invoke( this, new TicketRealtimeNotificationEventArgs(notification)); } private void StopCore() { EventSource source; lock (_sync) { _running = false; source = _eventSource; _eventSource = null; } if (source == null) return; source.Opened -= OnOpened; source.Closed -= OnClosed; source.MessageReceived -= OnMessageReceived; source.Error -= OnError; source.Close(); } private void SetState(TicketRealtimeState state, string message) { State = state; StateChanged?.Invoke( this, new TicketRealtimeStateChangedEventArgs(state, message)); } //private static Uri BuildEventsUri(TicketClientOptions options) //{ // var appIds = options.AppIds // .Where(a => !string.IsNullOrWhiteSpace(a)) // .Distinct(StringComparer.Ordinal) // .Select(a => "appId=" + Uri.EscapeDataString(a.Trim())); // var query = new[] // { // "purchaseAccount=" + Uri.EscapeDataString(options.UserName) // }.Concat(appIds); // return new Uri( // TicketApiService.ServerUrl // + EventsPath // + "?" // + string.Join("&", query), // UriKind.Absolute); //} private static Configuration BuildEventsUri(TicketClientOptions options) { var appIds = options.AppIds .Where(a => !string.IsNullOrWhiteSpace(a)) .Distinct(StringComparer.Ordinal) .Select(a => Uri.EscapeDataString(a.Trim())); var query = new[] { "purchaseAccount=" + Uri.EscapeDataString(options.UserName) }; var uri = new Uri( TicketApiService.ServerUrl + EventsPath + "?" + string.Join("&", query), UriKind.Absolute); ConfigurationBuilder builder = Configuration.Builder(uri); builder.RequestHeader("X-AppIds", $"[\"{string.Join("\",\"", appIds)}\"]"); builder.Logger(NLogger.Instance); //builder.LastEventId(""); //var proxy = new WebProxy("http://127.0.0.1:8081"); // //var handler = new HttpClientHandler //{ // Proxy = proxy, // UseProxy = true //}; //var client = new HttpClient(handler); //builder.HttpClient(client); builder._responseStartTimeout = TimeSpan.FromSeconds(30); return builder.Build(); } }