OA2023/OASystem/OASystem.Infrastructure/Tools/SseAlchemyHelper.cs

using Microsoft.AspNetCore.Http.Features;
using Newtonsoft.Json;
using Newtonsoft.Json.Serialization;

/// <summary>
/// SSE 流式数据助手类，提供初始化 SSE、发送数据包和结束流的方法
/// </summary>
public static class SseAlchemyHelper
{
    private static readonly JsonSerializerSettings _jsonSettings = new()
    {
        ContractResolver = new CamelCasePropertyNamesContractResolver(),
        DateTimeZoneHandling = DateTimeZoneHandling.Local,
        NullValueHandling = NullValueHandling.Ignore
    };

    /// <summary>
    /// 初始化 SSE
    /// </summary>
    public static void InitializeSse(this HttpContext context)
    {
        var syncIOFeature = context.Features.Get<IHttpBodyControlFeature>();
        if (syncIOFeature != null) syncIOFeature.AllowSynchronousIO = true;

        var response = context.Response;
        response.Headers.Append("Content-Type", "text/event-stream");
        response.Headers.Append("Cache-Control", "no-cache");
        response.Headers.Append("Connection", "keep-alive");
        response.Headers.Append("X-Accel-Buffering", "no");
    }

    /// <summary>
    /// 发送流数据包
    /// </summary>
    public static async Task SendSseStepAsync(
        this HttpContext context,
        int progress,
        string message,
        object? data = null)
    {
        // 检查客户端是否已断开连接
        if (context.RequestAborted.IsCancellationRequested) return;

        var payload = JsonConvert.SerializeObject(new { progress, message, data }, _jsonSettings);
        string sseFormattedData = $"data: {payload}\n\n";
        byte[] bytes = Encoding.UTF8.GetBytes(sseFormattedData);

        try
        {
            // 写入并立即冲刷到客户端
            await context.Response.Body.WriteAsync(bytes, 0, bytes.Length, context.RequestAborted);
            await context.Response.Body.FlushAsync(context.RequestAborted);
        }
        catch (OperationCanceledException) { /* 客户端取消，优雅退出 */ }
        catch (Exception) { /* 忽略其他写入异常 */ }
    }

    /// <summary>
    /// 结束 SSE 流
    /// </summary>
    public static async Task FinalizeSseAsync(this HttpContext context)
    {
        if (!context.RequestAborted.IsCancellationRequested)
        {
            await context.Response.Body.FlushAsync();
        }
    }
}