29WxOrder/OrderSupportTicket/Services/EventSourceTicketRealtimeService.cs

using LaunchDarkly.EventSource;
using OrderSupportTicket.Models;
using System;
using System.IO;
using System.Linq;
using System.Runtime.Serialization;
using System.Runtime.Serialization.Json;
using System.Text;
using System.Threading.Tasks;

namespace OrderSupportTicket.Services;

/// <summary>
/// 基于 LaunchDarkly.EventSource 的 SSE 客户端。
/// net462 可使用；断线自动退避重连，并携带服务端事件 ID 进行补偿。
/// </summary>
public sealed class EventSourceTicketRealtimeService : ITicketRealtimeService
{
    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 void Start(TicketClientOptions options)
    {
        if (_disposed)
            throw new ObjectDisposedException(nameof(EventSourceTicketRealtimeService));
        if (options == null)
            throw new ArgumentNullException(nameof(options));

        TicketApiService.ValidateOptions(options);
        StopCore();

        var uri = BuildEventsUri(options);
        var source = new EventSource(uri);
        source.Opened += OnOpened;
        source.Closed += OnClosed;
        source.MessageReceived += OnMessageReceived;
        source.Error += OnError;

        lock (_sync)
        {
            _running = true;
            _eventSource = source;
        }

        SetState(TicketRealtimeState.Connecting, "连接中…");
        _ = RunAsync(source);
    }

    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().ConfigureAwait(false);
            lock (_sync)
            {
                if (!_running || !ReferenceEquals(_eventSource, source))
                    return;
            }
            SetState(TicketRealtimeState.Reconnecting, "连接已关闭，等待重连…");
        }
        catch (Exception ex)
        {
            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
        {
            var serializer = new DataContractJsonSerializer(
                typeof(TicketRealtimeNotification));
            using (var stream = new MemoryStream(Encoding.UTF8.GetBytes(data)))
            {
                notification = (TicketRealtimeNotification)serializer.ReadObject(stream);
            }
        }
        catch (SerializationException)
        {
            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);
    }
}