29WxOrder/OrderSupportTicket/Services/EventSourceTicketRealtimeService.cs

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;

/// <summary>
/// 基于 LaunchDarkly.EventSource 的 SSE 客户端。
/// net462 可使用；断线自动退避重连，并携带服务端事件 ID 进行补偿。
/// </summary>
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<TicketRealtimeNotificationEventArgs> NotificationReceived;

	public event EventHandler<TicketRealtimeStateChangedEventArgs> StateChanged;

	/// <summary>
	/// 当前 SSE 连接状态。
	/// </summary>
	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<TicketRealtimeNotification>(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();
	}
}