SSE (Server-Sent Events) в ASP.NET Core и .NET 10

Сервер ---> Клиент  // SSE: односторонняя связь
 
Сервер <--> Клиент  // WebSocket: двусторонняя связь

event: order
id: 12345
data: {"product": "Книга", "price": 500}

Polling (5 сек):   720 запросов/час, ~1.2 МБ/час
Long-polling:      120 запросов/час, ~0.4 МБ/час
SSE:               1-2 запроса/час,  ~0.2 МБ/час

Polling:      CPU: 75-85%, RAM: 1.2 GB
Long-polling: CPU: 40-50%, RAM: 1.8 GB
SSE:          CPU: 15-25%, RAM: 0.9 GB

Polling (5 сек):  2.5 сек (в среднем)
Long-polling:     0.1-0.2 сек
SSE:              0.1-0.2 сек

Polling:      DOM Events: ~720/час, JS CPU: 3-5%
Long-polling: DOM Events: ~120/час, JS CPU: 1-2%
SSE:          DOM Events: ~по числу реальных событий, JS CPU: <1%

Polling (5 сек): -12% заряда батареи
Long-polling:    -7% заряда батареи
SSE:             -4% заряда батареи

app.MapGet("/events", (CancellationToken cancellationToken) =>
    TypedResults.ServerSentEvents(
        GetEventsAsync(cancellationToken),
        eventType: "message"
    )
);
 
async IAsyncEnumerable<string> GetEventsAsync(
    [EnumeratorCancellation] CancellationToken cancellationToken)
{
    while (!cancellationToken.IsCancellationRequested)
    {
        yield return $"Время на сервере: {DateTime.Now}";
        await Task.Delay(1000, cancellationToken);
    }
}

public class OrderService
{
    private readonly Subject<Order> _orderSubject = new();
 
    public void AddOrder(Order order)
    {
        _orderSubject.OnNext(order);
    }
 
    public IAsyncEnumerable<string> GetOrderUpdatesAsync(
        CancellationToken cancellationToken)
    {
        return _orderSubject
            .Select(order => JsonSerializer.Serialize(order))
            .ToAsyncEnumerable()
            .WithCancellation(cancellationToken);
    }
}

app.MapGet("/orders/updates", (OrderService orderService, CancellationToken cancellationToken) =>
    TypedResults.ServerSentEvents(
        orderService.GetOrderUpdatesAsync(cancellationToken),
        eventType: "order"
    )
);

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

public static IResult ServerSentEventsWithTimeout<T>(
    this IResultExtensions extensions,
    IAsyncEnumerable<T> source,
    string? eventType = null,
    Func<T, string?>? eventId = null,
    TimeSpan timeout = default)
{
    timeout = timeout == default ? TimeSpan.FromMinutes(2) : timeout;
    
    return new CustomServerSentEventsResult<T>(
        source, 
        eventType, 
        eventId, 
        timeout);
}

public static class ServiceCollectionExtensions
{
    public static IServiceCollection AddServerSentEvents(
        this IServiceCollection services)
    {
        services.AddSingleton<IEventSourceService, EventSourceService>();
        services.AddHostedService<EventBroadcastService>();
        return services;
    }
}

public class EventSourceService : IEventSourceService
{
    private readonly ConcurrentDictionary<string, Channel<string>> _channels 
        = new();
    
    public IAsyncEnumerable<string> Subscribe(string channelName, 
        CancellationToken cancellationToken)
    {
        var channel = _channels.GetOrAdd(channelName, 
            _ => Channel.CreateUnbounded<string>());
        
        return channel.Reader.ReadAllAsync(cancellationToken);
    }
    
    public ValueTask PublishAsync(string channelName, string message)
    {
        if (_channels.TryGetValue(channelName, out var channel))
        {
            return channel.Writer.WriteAsync(message);
        }
        
        return ValueTask.CompletedTask;
    }
}

public class ServerSentEventsMiddleware
{
    private readonly RequestDelegate _next;
    
    public ServerSentEventsMiddleware(RequestDelegate next)
    {
        _next = next;
    }
    
    public async Task InvokeAsync(HttpContext context, 
        IEventSourceService eventSource)
    {
        if (!context.Request.Path.StartsWithSegments("/sse"))
        {
            await _next(context);
            return;
        }
        
        string channelName = context.Request.Query["channel"];
        if (string.IsNullOrEmpty(channelName))
        {
            context.Response.StatusCode = 400;
            return;
        }
        
        context.Response.Headers.Add("Content-Type", "text/event-stream");
        context.Response.Headers.Add("Cache-Control", "no-cache");
        context.Response.Headers.Add("Connection", "keep-alive");
        
        var cancellationToken = context.RequestAborted;
        
        await foreach (var message in eventSource
            .Subscribe(channelName, cancellationToken)
            .WithCancellation(cancellationToken))
        {
            await context.Response.WriteAsync($"data: {message}\r\n\r\n",
                cancellationToken);
            await context.Response.Body.FlushAsync(cancellationToken);
        }
    }
}

app.UseMiddleware<ServerSentEventsMiddleware>();

app.Use(async (context, next) =>
{
    context.Response.Headers.Add("X-Content-Type-Options", "nosniff");
    context.Features.Get<IHttpMaxRequestBodySizeFeature>()
        ?.MaxRequestBodySize = null;
    
    context.Response.BufferOutput = false;
    await next();
});

builder.WebHost.ConfigureKestrel(options =>
{
    options.Limits.KeepAliveTimeout = TimeSpan.FromMinutes(2);
    options.Limits.RequestHeadersTimeout = TimeSpan.FromMinutes(1);
    options.Limits.MaxConcurrentConnections = 10000;
});

http {
    proxy_buffering off;
    proxy_read_timeout 3600s;
    proxy_connect_timeout 3600s;
    proxy_send_timeout 3600s;
    
    server {
        location /sse/ {
            proxy_pass [url]http://backend;[/url]
            proxy_http_version 1.1;
            proxy_set_header Connection "";
        }
    }
}

public class EventHub
{
    private readonly ConcurrentDictionary<string, EventChannel> _channels = new();
    
    public EventChannel GetOrCreateChannel(string name)
    {
        return _channels.GetOrAdd(name, _ => new EventChannel(name));
    }
}
 
public class EventChannel
{
    private readonly string _name;
    private readonly Channel<object> _channel = Channel.CreateUnbounded<object>(
        new UnboundedChannelOptions { SingleReader = false, SingleWriter = false });
    
    public EventChannel(string name)
    {
        _name = name;
    }
    
    public async ValueTask PublishAsync(object data)
    {
        await _channel.Writer.WriteAsync(data);
    }
    
    public IAsyncEnumerable<string> Subscribe(CancellationToken cancellationToken)
    {
        return _channel.Reader
            .ReadAllAsync(cancellationToken)
            .Select(data => JsonSerializer.Serialize(data));
    }
}

app.MapGet("/stream", async (HttpContext context, 
    IEventSourceService service, 
    CancellationToken cancellationToken) =>
{
    try
    {
        return TypedResults.ServerSentEvents(
            service.GetEvents(cancellationToken),
            eventType: "update"
        );
    }
    catch (OperationCanceledException)
    {
        // Клиент отключился, просто логируем
        logger.LogInformation("Клиент отключился");
        return Results.Empty;
    }
    catch (Exception ex)
    {
        logger.LogError(ex, "Ошибка при обработке SSE-соединения");
        return Results.Problem("Внутренняя ошибка сервера");
    }
});

public class SseConnectionMetrics
{
    private readonly Counter<int> _activeConnections;
    private readonly Counter<int> _totalConnections;
    private readonly Counter<int> _messagesSent;
    
    public SseConnectionMetrics(IMeterFactory meterFactory)
    {
        var meter = meterFactory.Create("SseMetrics");
        _activeConnections = meter.CreateCounter<int>("active_connections");
        _totalConnections = meter.CreateCounter<int>("total_connections");
        _messagesSent = meter.CreateCounter<int>("messages_sent");
    }
    
    public void ConnectionOpened() 
    {
        _activeConnections.Add(1);
        _totalConnections.Add(1);
    }
    
    public void ConnectionClosed() 
    {
        _activeConnections.Add(-1);
    }
    
    public void MessageSent() 
    {
        _messagesSent.Add(1);
    }
}

public class EventNotifier : INotificationHandler<DomainEventNotification>
{
    private readonly IEventHub _eventHub;
    
    public EventNotifier(IEventHub eventHub)
    {
        _eventHub = eventHub;
    }
    
    public async Task Handle(DomainEventNotification notification, 
        CancellationToken cancellationToken)
    {
        var channelName = notification.Event.GetType().Name;
        var channel = _eventHub.GetOrCreateChannel(channelName);
        await channel.PublishAsync(notification.Event);
    }
}

app.MapGet("/secure-stream", 
    [Authorize(Roles = "Premium")] 
    (CancellationToken token) =>
    TypedResults.ServerSentEvents(GetPremiumEvents(token)))
.RequireAuthorization();

var connectionCounter = new SemaphoreSlim(1000); // макс. 1000 соединений
 
app.MapGet("/limited-stream", async (CancellationToken token) =>
{
    if (!await connectionCounter.WaitAsync(0))
        return Results.StatusCode(503); // Сервис перегружен
    
    try
    {
        return TypedResults.ServerSentEvents(GetEvents(token));
    }
    finally
    {
        connectionCounter.Release();
    }
});

[AttributeUsage(AttributeTargets.Method)]
public class ServerSentEventAttribute : Attribute
{
    public string? EventType { get; set; }
    public int BufferSize { get; set; } = 1024;
    public TimeSpan Timeout { get; set; } = TimeSpan.FromMinutes(2);
    public bool IncludeEventId { get; set; } = true;
}

public static class ServerSentEventsExtensions
{
    public static IEndpointRouteBuilder MapServerSentEvents(
        this IEndpointRouteBuilder endpoints, 
        Assembly assembly)
    {
        var methods = assembly.GetTypes()
            .SelectMany(t => t.GetMethods())
            .Where(m => m.GetCustomAttribute<ServerSentEventAttribute>() != null)
            .ToList();
            
        foreach (var method in methods)
        {
            var attribute = method.GetCustomAttribute<ServerSentEventAttribute>();
            var path = method.GetCustomAttribute<RouteAttribute>()?.Template 
                       ?? $"/sse/{method.Name.ToLowerInvariant()}";
                       
            endpoints.MapGet(path, async context => 
            {
                // Настраиваем заголовки
                context.Response.Headers.Add("Content-Type", "text/event-stream");
                context.Response.Headers.Add("Cache-Control", "no-cache");
                context.Response.Headers.Add("Connection", "keep-alive");
                
                // Получаем сервис и вызываем метод
                var serviceType = method.DeclaringType!;
                var service = context.RequestServices.GetRequiredService(serviceType);
                
                // Предполагаем, что метод возвращает IAsyncEnumerable<string>
                var enumerable = (IAsyncEnumerable<string>)method.Invoke(
                    service, new object[] { context.RequestAborted })!;
                
                var cancellationToken = context.RequestAborted;
                
                // Отправляем события
                await foreach (var item in enumerable.WithCancellation(cancellationToken))
                {
                    var builder = new StringBuilder();
                    if (!string.IsNullOrEmpty(attribute!.EventType))
                    {
                        builder.AppendLine($"event: {attribute.EventType}");
                    }
                    
                    if (attribute!.IncludeEventId)
                    {
                        builder.AppendLine($"id: {Guid.NewGuid()}");
                    }
                    
                    builder.AppendLine($"data: {item}");
                    builder.AppendLine();
                    
                    await context.Response.WriteAsync(
                        builder.ToString(), cancellationToken);
                    await context.Response.Body.FlushAsync(cancellationToken);
                }
            });
        }
        
        return endpoints;
    }
}

app.MapServerSentEvents(typeof(Program).Assembly);

public class StockService
{
    [ServerSentEvent(EventType = "stock")]
    [Route("/stocks/updates")]
    public async IAsyncEnumerable<string> GetStockUpdates(
        [EnumeratorCancellation] CancellationToken token)
    {
        while (!token.IsCancellationRequested)
        {
            yield return JsonSerializer.Serialize(
                new { Symbol = "MSFT", Price = Random.Shared.Next(250, 350) });
            await Task.Delay(1000, token);
        }
    }
}

[AttributeUsage(AttributeTargets.Method)]
public class SseRetryAttribute : Attribute
{
    public int RetryMilliseconds { get; }
    
    public SseRetryAttribute(int retryMilliseconds)
    {
        RetryMilliseconds = retryMilliseconds;
    }
}

// В методе MapServerSentEvents добавляем:
var retryAttribute = method.GetCustomAttribute<SseRetryAttribute>();
if (retryAttribute != null)
{
    builder.AppendLine($"retry: {retryAttribute.RetryMilliseconds}");
}

public interface ISseController
{
    IAsyncEnumerable<string> GetEventStreamAsync(CancellationToken cancellationToken);
    string EventType { get; }
    int RetryInterval { get; }
}

public abstract class BaseSseController : ISseController
{
    public abstract string EventType { get; }
    public virtual int RetryInterval => 3000; // 3 секунды по умолчанию
    
    protected readonly ILogger _logger;
    
    protected BaseSseController(ILogger logger)
    {
        _logger = logger;
    }
    
    public abstract IAsyncEnumerable<string> GetEventStreamAsync(
        CancellationToken cancellationToken);
    
    protected ValueTask<string> SerializeEventDataAsync<T>(T data)
    {
        try 
        {
            return ValueTask.FromResult(JsonSerializer.Serialize(data));
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Ошибка сериализации данных события");
            throw;
        }
    }
}

public class OrderEventsController : BaseSseController
{
    private readonly IOrderEventService _orderService;
    
    public OrderEventsController(ILogger<OrderEventsController> logger, 
                               IOrderEventService orderService) 
        : base(logger)
    {
        _orderService = orderService;
    }
    
    public override string EventType => "order";
    
    public override async IAsyncEnumerable<string> GetEventStreamAsync(
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        await foreach (var order in _orderService.GetOrderUpdatesAsync(cancellationToken))
        {
            yield return await SerializeEventDataAsync(order);
        }
    }
}

app.MapGet("/sse/{controller}", async (string controller, 
                                     [FromServices] IServiceProvider services,
                                     HttpContext context) =>
{
    var controllerType = Type.GetType($"MyApp.Controllers.{controller}SseController");
    if (controllerType == null || !typeof(ISseController).IsAssignableFrom(controllerType))
    {
        return Results.NotFound();
    }
    
    var sseController = (ISseController)services.GetRequiredService(controllerType);
    
    return TypedResults.ServerSentEvents(
        sseController.GetEventStreamAsync(context.RequestAborted),
        eventType: sseController.EventType
    );
});

app.MapGet("/secure-stream", async (HttpContext context, CancellationToken token) =>
{
    // Проверка наличия необходимых заголовков
    if (!context.Request.Headers.TryGetValue("X-Client-Id", out var clientId))
    {
        return Results.BadRequest("Отсутствует идентификатор клиента");
    }
    
    // Проверка лимитов соединений для конкретного клиента
    if (!await _connectionLimiter.TryAcquireAsync(clientId))
    {
        return Results.StatusCode(429); // Too Many Requests
    }
    
    // Если все проверки пройдены, запускаем SSE
    return TypedResults.ServerSentEvents(GetSecureEvents(clientId, token));
});

public class OriginValidationMiddleware
{
    private readonly RequestDelegate _next;
    private readonly HashSet<string> _allowedOrigins;
 
    public OriginValidationMiddleware(RequestDelegate next, IConfiguration config)
    {
        _next = next;
        _allowedOrigins = new HashSet<string>(
            config.GetSection("AllowedOrigins").Get<string[]>() ?? Array.Empty<string>());
    }
 
    public async Task InvokeAsync(HttpContext context)
    {
        // Проверяем только SSE-запросы
        if (context.Request.Path.StartsWithSegments("/sse"))
        {
            var origin = context.Request.Headers.Origin.ToString();
            
            if (string.IsNullOrEmpty(origin) || !_allowedOrigins.Contains(origin))
            {
                context.Response.StatusCode = 403;
                return;
            }
        }
 
        await _next(context);
    }
}

app.UseMiddleware<OriginValidationMiddleware>();

app.MapGet("/sse-with-token", async (string token, CancellationToken ct) =>
{
    if (!await _tokenValidator.ValidateAndConsumeAsync(token))
    {
        return Results.Unauthorized();
    }
    
    return TypedResults.ServerSentEvents(GetEventStream(ct));
});

public class BoundedEventChannel<T>
{
    private readonly Channel<T> _channel;
    private readonly ChannelWriter<T> _writer;
    private readonly ChannelReader<T> _reader;
 
    public BoundedEventChannel(int capacity, 
                           BoundedChannelFullMode fullMode = BoundedChannelFullMode.Wait)
    {
        _channel = Channel.CreateBounded<T>(new BoundedChannelOptions(capacity)
        {
            FullMode = fullMode,
            SingleWriter = false,
            SingleReader = false
        });
        
        _writer = _channel.Writer;
        _reader = _channel.Reader;
    }
 
    public ValueTask WriteAsync(T item, CancellationToken ct = default)
    {
        return _writer.WriteAsync(item, ct);
    }
 
    public IAsyncEnumerable<T> ReadAllAsync(CancellationToken ct = default)
    {
        return _reader.ReadAllAsync(ct);
    }
}

if (isHighPriority && _channel.Writer.TryWrite(item))
{
    return; // Высокоприоритетные события пишем напрямую
}
else
{
    // Для обычных событий используем WriteAsync с таймаутом
    using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(100));
    await _channel.Writer.WriteAsync(item, cts.Token);
}

app.Use(async (context, next) =>
{
    context.Response.BufferOutput = false;
    await next();
});

builder.WebHost.ConfigureKestrel(options =>
{
    options.Limits.MaxResponseBufferSize = 64 * 1024; // 64 KB
});

public class SseAuditMiddleware
{
    private readonly RequestDelegate _next;
    private readonly ILogger<SseAuditMiddleware> _logger;
 
    public SseAuditMiddleware(RequestDelegate next, ILogger<SseAuditMiddleware> logger)
    {
        _next = next;
        _logger = logger;
    }
 
    public async Task InvokeAsync(HttpContext context)
    {
        if (context.Request.Path.StartsWithSegments("/sse"))
        {
            var clientIp = context.Connection.RemoteIpAddress?.ToString() ?? "unknown";
            var userAgent = context.Request.Headers.UserAgent.ToString();
            var userId = context.User.Identity?.IsAuthenticated == true 
                ? context.User.FindFirst(ClaimTypes.NameIdentifier)?.Value 
                : "anonymous";
            
            using var loggingScope = _logger.BeginScope(new Dictionary<string, object>
            {
                ["ClientIp"] = clientIp,
                ["UserAgent"] = userAgent,
                ["UserId"] = userId,
                ["ConnectionId"] = context.Connection.Id
            });
            
            _logger.LogInformation("SSE connection established");
            
            var originalBodyStream = context.Response.Body;
            using var responseBody = new MemoryStream();
            context.Response.Body = responseBody;
            
            var startTime = DateTime.UtcNow;
            var eventCounter = 0;
            
            context.Response.OnStarting(() => 
            {
                context.Response.Headers["X-SSE-Audit-Id"] = Guid.NewGuid().ToString();
                return Task.CompletedTask;
            });
            
            try 
            {
                await _next(context);
            }
            catch (Exception ex) 
            {
                _logger.LogError(ex, "Error in SSE connection");
                throw;
            }
            finally 
            {
                var duration = DateTime.UtcNow - startTime;
                _logger.LogInformation(
                    "SSE connection closed after {Duration}. Events sent: {EventCount}", 
                    duration, eventCounter);
            }
        }
        else 
        {
            await _next(context);
        }
    }
}

app.UseMiddleware<SseAuditMiddleware>();

private readonly Counter _sseConnectionsTotal;
private readonly Gauge _sseConnectionsActive;
private readonly Counter _sseEventsTotal;
 
public SseAuditMiddleware(RequestDelegate next, 
                          ILogger<SseAuditMiddleware> logger,
                          IMetricsFactory metrics)
{
    _next = next;
    _logger = logger;
    
    _sseConnectionsTotal = metrics.CreateCounter("sse_connections_total", 
        "Total number of SSE connections");
    _sseConnectionsActive = metrics.CreateGauge("sse_connections_active", 
        "Current active SSE connections");
    _sseEventsTotal = metrics.CreateCounter("sse_events_total", 
        "Total SSE events sent");
}

public class GeoLocationFormatter : ISseFormatter<GeoLocation>
{
public string Format(GeoLocation location)
{
    // Оптимизированный формат для передачи гео-данных
    return $"{location.Latitude:F6},{location.Longitude:F6},{location.Accuracy:F2}";
}
}

public interface ISseFormatter<T>
{
string Format(T data);
}

public class SseFormatterFactory
{
private readonly Dictionary<Type, object> _formatters = new();
 
public void Register<T>(ISseFormatter<T> formatter)
{
    _formatters[typeof(T)] = formatter;
}
 
public string Format<T>(T data)
{
    if (_formatters.TryGetValue(typeof(T), out var formatterObj))
    {
        var formatter = (ISseFormatter<T>)formatterObj;
        return formatter.Format(data);
    }
    
    // Fallback на стандартную JSON-сериализацию
    return JsonSerializer.Serialize(data);
}
}

public class LocationEventController : BaseSseController
{
private readonly ILocationService _locationService;
private readonly SseFormatterFactory _formatterFactory;
 
public LocationEventController(
    ILocationService locationService,
    SseFormatterFactory formatterFactory)
{
    _locationService = locationService;
    _formatterFactory = formatterFactory;
}
 
public override async IAsyncEnumerable<string> GetEventStreamAsync(
    [EnumeratorCancellation] CancellationToken token)
{
    await foreach (var location in _locationService.GetLocationsAsync(token))
    {
        // Используем специальный форматтер вместо JSON
        yield return _formatterFactory.Format(location);
    }
}
}

public class SseConnectionPool
{
    private readonly ConcurrentBag<SseConnection> _availableConnections = new();
    private readonly SemaphoreSlim _poolLock = new(1, 1);
    private int _totalConnections;
    private readonly int _maxPoolSize;
 
    public SseConnectionPool(int maxPoolSize = 1000)
    {
        _maxPoolSize = maxPoolSize;
    }
 
    public async Task<SseConnection?> AcquireAsync(CancellationToken token)
    {
        if (_availableConnections.TryTake(out var connection))
        {
            return connection;
        }
 
        await _poolLock.WaitAsync(token);
        try
        {
            // Повторная проверка после получения блокировки
            if (_availableConnections.TryTake(out connection))
            {
                return connection;
            }
 
            if (_totalConnections < _maxPoolSize)
            {
                connection = new SseConnection();
                Interlocked.Increment(ref _totalConnections);
                return connection;
            }
 
            // Пул исчерпан, придется подождать
            return null;
        }
        finally
        {
            _poolLock.Release();
        }
    }
 
    public void Release(SseConnection connection)
    {
        connection.Reset();
        _availableConnections.Add(connection);
    }
}

public class SseConnection
{
    private Channel<string>? _channel;
    public ChannelWriter<string> Writer => _channel!.Writer;
    public ChannelReader<string> Reader => _channel!.Reader;
    
    public SseConnection()
    {
        Reset();
    }
    
    public void Reset()
    {
        _channel = Channel.CreateUnbounded<string>(
            new UnboundedChannelOptions { SingleReader = true, SingleWriter = false });
    }
}

app.MapGet("/pooled-sse", async (HttpContext context, 
                            SseConnectionPool pool,
                            CancellationToken token) =>
{
    var connection = await pool.AcquireAsync(token);
    if (connection == null)
    {
        return Results.StatusCode(503); // Сервис перегружен
    }
    
    try
    {
        // Запускаем фоновую задачу для публикации событий
        _ = Task.Run(async () => 
        {
            try 
            {
                for (int i = 0; !token.IsCancellationRequested; i++)
                {
                    await connection.Writer.WriteAsync($"Event {i}", token);
                    await Task.Delay(1000, token);
                }
            }
            catch (OperationCanceledException) { /* Клиент отключился */ }
        }, token);
        
        return TypedResults.ServerSentEvents(
            connection.Reader.ReadAllAsync(token),
            eventType: "update"
        );
    }
    finally
    {
        if (token.IsCancellationRequested)
        {
            pool.Release(connection);
        }
    }
});

builder.WebHost.ConfigureKestrel(options =>
{
    options.ListenAnyIP(5001, listenOptions =>
    {
        listenOptions.Protocols = HttpProtocols.Http2;
        // Для HTTPS
        listenOptions.UseHttps();
    });
});

builder.Services.Configure<KestrelServerOptions>(options =>
{
    options.Limits.MaxConcurrentConnections = 10000;
    options.Limits.MaxConcurrentUpgradedConnections = 10000;
    options.Limits.Http2.MaxStreamsPerConnection = 100;
    options.Limits.Http2.InitialConnectionWindowSize = 1048576; // 1 MB
    options.Limits.Http2.InitialStreamWindowSize = 262144; // 256 KB
});

options.Limits.Http2.KeepAlivePingTimeout = TimeSpan.FromSeconds(20);
options.Limits.Http2.KeepAlivePingDelay = TimeSpan.FromSeconds(30);

public class HeartbeatService
{
private readonly TimeSpan _interval = TimeSpan.FromSeconds(30);
 
public async IAsyncEnumerable<string> GenerateHeartbeats(
    [EnumeratorCancellation] CancellationToken cancellationToken)
{
    while (!cancellationToken.IsCancellationRequested)
    {
        yield return ":heartbeat"; // Комментарий в SSE начинается с двоеточия
        await Task.Delay(_interval, cancellationToken);
    }
}
}

app.MapGet("/sse-with-heartbeat", (HeartbeatService heartbeat, 
                                  EventService events,
                                  CancellationToken token) =>
{
    var combinedStream = events.GetEvents(token)
        .Merge(heartbeat.GenerateHeartbeats(token));
        
    return TypedResults.ServerSentEvents(combinedStream);
});

public static async IAsyncEnumerable<T> Merge<T>(
    this IAsyncEnumerable<T> first,
    IAsyncEnumerable<T> second,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    var tasks = new List<Task<(bool HasValue, T Value)>>();
    
    using var enum1 = first.GetAsyncEnumerator(cancellationToken);
    using var enum2 = second.GetAsyncEnumerator(cancellationToken);
    
    tasks.Add(GetNextAsync(enum1, cancellationToken));
    tasks.Add(GetNextAsync(enum2, cancellationToken));
    
    while (tasks.Count > 0 && !cancellationToken.IsCancellationRequested)
    {
        var completed = await Task.WhenAny(tasks);
        tasks.Remove(completed);
        
        var (hasValue, value) = await completed;
        if (hasValue)
        {
            yield return value;
            
            if (completed == tasks[0])
                tasks.Add(GetNextAsync(enum1, cancellationToken));
            else
                tasks.Add(GetNextAsync(enum2, cancellationToken));
        }
    }
}
 
private static async Task<(bool HasValue, T Value)> GetNextAsync<T>(
    IAsyncEnumerator<T> enumerator, 
    CancellationToken cancellationToken)
{
    try
    {
        if (await enumerator.MoveNextAsync())
            return (true, enumerator.Current);
    }
    catch (OperationCanceledException) { }
    
    return (false, default!);
}

private async Task<bool> IsConnectionAlive(HttpContext context, 
                                          CancellationToken token)
{
    try
    {
        await context.Response.WriteAsync(":ping\n\n", token);
        await context.Response.Body.FlushAsync(token);
        return true;
    }
    catch
    {
        return false;
    }
}

builder.Services.AddCors(options =>
{
options.AddPolicy("SsePolicy", builder =>
{
    builder
        .WithOrigins("https://trusted-client.com")
        .AllowCredentials()
        .WithHeaders("X-Requested-With", "Content-Type", "Authorization")
        .WithExposedHeaders("X-SSE-Event-Id", "X-SSE-Error")
        .SetPreflightMaxAge(TimeSpan.FromHours(1));
});
});

app.MapGet("/events", (CancellationToken token) =>
    TypedResults.ServerSentEvents(GetEvents(token)))
.RequireCors("SsePolicy");

app.UseCors(builder =>
{
builder
    .AllowAnyMethod()
    .AllowCredentials()
    .WithHeaders("Content-Type", "Accept", "X-Requested-With")
    .SetIsOriginAllowed(origin => 
    {
        var host = new Uri(origin).Host;
        return host.EndsWith("mycompany.com") || 
               host == "localhost";
    });
});

# NGINX configuration
location /sse/ {
    proxy_pass [url]http://backend;[/url]
    
    # Preserve CORS headers
    proxy_set_header Origin $http_origin;
    proxy_pass_header Access-Control-Allow-Origin;
    proxy_pass_header Access-Control-Allow-Methods;
    proxy_pass_header Access-Control-Allow-Headers;
    proxy_pass_header Access-Control-Allow-Credentials;
    proxy_pass_header Access-Control-Max-Age;
}

public interface IEventPublisher<T>
{
    ValueTask PublishAsync(T eventData);
}
 
public interface IEventSubscriber<T>
{
    IAsyncEnumerable<T> SubscribeAsync(CancellationToken cancellationToken);
}
 
// Реализация для SSE
public class SseEventBroker<T> : IEventPublisher<T>, IEventSubscriber<T>
{
    private readonly Channel<T> _channel = Channel.CreateUnbounded<T>();
 
    public ValueTask PublishAsync(T eventData)
    {
        return _channel.Writer.WriteAsync(eventData);
    }
 
    public IAsyncEnumerable<T> SubscribeAsync(CancellationToken cancellationToken)
    {
        return _channel.Reader.ReadAllAsync(cancellationToken);
    }
}

public class EventAggregator<T>
{
    private readonly List<IEventSubscriber<T>> _sources = new();
    
    public void AddSource(IEventSubscriber<T> source)
    {
        _sources.Add(source);
    }
    
    public async IAsyncEnumerable<T> SubscribeAll(
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        var streams = _sources.Select(s => s.SubscribeAsync(cancellationToken)).ToList();
        var enumerators = streams.Select(s => s.GetAsyncEnumerator(cancellationToken)).ToList();
        
        try
        {
            // Упрощенная реализация, в реальном коде нужен более сложный механизм
            while (!cancellationToken.IsCancellationRequested)
            {
                foreach (var enumerator in enumerators)
                {
                    if (await enumerator.MoveNextAsync())
                        yield return enumerator.Current;
                }
                await Task.Delay(10, cancellationToken);
            }
        }
        finally
        {
            foreach (var enumerator in enumerators)
                await enumerator.DisposeAsync();
        }
    }
}

public static class EventFilterExtensions
{
    public static IAsyncEnumerable<T> WhereAsync<T>(
        this IAsyncEnumerable<T> source, 
        Func<T, bool> predicate)
    {
        return source.Where(predicate);
    }
}

app.MapGet("/sse", async (HttpContext context) =>
{
    var cancellationToken = context.RequestAborted;
    // Используем этот токен для всех асинхронных операций
    await foreach (var item in GetDataStream().WithCancellation(cancellationToken))
    {
        // ...
    }
});

private bool IsConnectionActive(HttpContext context)
{
    try
    {
        context.Response.Body.WriteAsync(
            new byte[] { (byte)':' }, 0, 1, context.RequestAborted);
        return true;
    }
    catch
    {
        return false;
    }
}

private readonly ConcurrentDictionary<string, ClientState> _clients = new();
 
app.MapGet("/sse/connect", (HttpContext context) =>
{
    var connectionId = Guid.NewGuid().ToString();
    var state = new ClientState
    {
        UserId = context.User.FindFirst(ClaimTypes.NameIdentifier)?.Value,
        ConnectedAt = DateTime.UtcNow
    };
    
    _clients[connectionId] = state;
    
    context.Response.OnCompleted(() => 
    {
        _clients.TryRemove(connectionId, out _);
        return Task.CompletedTask;
    });
    
    // ...
});

public async Task ProcessSseConnection(HttpContext context)
{
    var subscription = new Subscription();
    try
    {
        // Работа с соединением
    }
    finally
    {
        await subscription.DisposeAsync();
    }
}

app.MapGet("/secure-events", [Authorize] (CancellationToken token) =>
TypedResults.ServerSentEvents(GetSecureEvents(token)))
.RequireAuthorization();

public class TokenValidationMiddleware
{
    private readonly RequestDelegate _next;
    private readonly ITokenValidator _tokenValidator;
 
    public TokenValidationMiddleware(RequestDelegate next, 
                                   ITokenValidator tokenValidator)
    {
        _next = next;
        _tokenValidator = tokenValidator;
    }
 
    public async Task InvokeAsync(HttpContext context)
    {
        if (context.Request.Path.StartsWithSegments("/sse"))
        {
            // Проверяем токен каждые 5 минут
            var validationTask = StartPeriodicValidation(context);
            try 
            {
                await _next(context);
            }
            finally 
            {
                await validationTask;
            }
        }
        else
        {
            await _next(context);
        }
    }
 
    private async Task StartPeriodicValidation(HttpContext context)
    {
        var cancellationToken = context.RequestAborted;
        while (!cancellationToken.IsCancellationRequested)
        {
            await Task.Delay(TimeSpan.FromMinutes(5), cancellationToken);
            
            var token = context.Request.Headers.Authorization.ToString().Replace("Bearer ", "");
            if (!await _tokenValidator.ValidateAsync(token))
            {
                // Насильственно разрываем соединение
                context.Abort();
                break;
            }
        }
    }
}

public class TokenExpirationService
{
    private readonly Channel<string> _expirationChannel = Channel.CreateUnbounded<string>();
    
    public void NotifyExpired(string tokenId)
    {
        _expirationChannel.Writer.TryWrite(tokenId);
    }
    
    public ChannelReader<string> GetExpirationReader()
    {
        return _expirationChannel.Reader;
    }
}

public class EventAuthorizationService
{
    private readonly Dictionary<string, string[]> _channelRoles = new()
    {
        ["orders"] = new[] { "Admin", "Sales" },
        ["system"] = new[] { "Admin" },
        ["analytics"] = new[] { "Admin", "Analyst", "Sales" }
    };
    
    public bool CanAccessChannel(ClaimsPrincipal user, string channelName)
    {
        if (!_channelRoles.TryGetValue(channelName, out var roles))
            return false; // Неизвестный канал
        
        return roles.Any(role => user.IsInRole(role));
    }
}

app.MapGet("/events/{channel}", (string channel, 
                              HttpContext context, 
                              EventAuthorizationService authService,
                              IEventService eventService,
                              CancellationToken token) =>
{
    if (!authService.CanAccessChannel(context.User, channel))
        return Results.Forbid();
    
    return TypedResults.ServerSentEvents(
        eventService.Subscribe(channel, token),
        eventType: channel
    );
});

builder.Services.AddAuthorization(options =>
{
    options.AddPolicy("CanReceiveUserEvents", policy =>
        policy.RequireAssertion(context => 
        {
            var userId = context.User.FindFirstValue(ClaimTypes.NameIdentifier);
            var requestedUser = context.Resource as string;
            
            return context.User.IsInRole("Admin") || 
                   userId == requestedUser;
        }));
});

app.MapGet("/users/{userId}/events", (string userId, 
                                  IUserEventService events,
                                  CancellationToken token) =>
TypedResults.ServerSentEvents(events.GetUserEvents(userId, token)))
.RequireAuthorization(policyName: "CanReceiveUserEvents");

public IAsyncEnumerable<OrderEvent> GetFilteredOrderEvents(
    ClaimsPrincipal user, 
    CancellationToken token)
{
    var region = user.FindFirstValue("Region");
    return _orderEvents
        .ReadAllAsync(token)
        .WhereAwait(async order => 
        {
            // Админы видят все заказы
            if (user.IsInRole("Admin"))
                return true;
                
            // Обычные пользователи видят заказы своего региона
            return order.Region == region;
        });
}

public class ServerSentEventsResult<T> : IResult
{
private readonly IAsyncEnumerable<T> _source;
private readonly string? _eventType;
private readonly Func<T, string?>? _eventIdProvider;
 
public async Task ExecuteAsync(HttpContext httpContext)
{
    // Настройка HTTP-заголовков
    httpContext.Response.ContentType = "text/event-stream";
    httpContext.Response.Headers.CacheControl = "no-cache";
    
    var cancellationToken = httpContext.RequestAborted;
    
    // Перебор асинхронной последовательности
    await foreach (var item in _source.WithCancellation(cancellationToken))
    {
        // Форматирование SSE-события
        var eventBuilder = new StringBuilder();
        
        if (_eventType is not null)
            eventBuilder.AppendLine($"event: {_eventType}");
            
        if (_eventIdProvider is not null)
        {
            var id = _eventIdProvider(item);
            if (id is not null)
                eventBuilder.AppendLine($"id: {id}");
        }
        
        var data = item?.ToString() ?? string.Empty;
        // Данные могут содержать переносы строк, 
        // поэтому каждую строку нужно префиксить "data: "
        foreach (var line in data.Split('\n'))
            eventBuilder.AppendLine($"data: {line}");
            
        eventBuilder.AppendLine();
        
        await httpContext.Response.WriteAsync(
            eventBuilder.ToString(), cancellationToken);
        await httpContext.Response.Body.FlushAsync(cancellationToken);
    }
}
}

public async IAsyncEnumerable<string> GetLargeDatasetAsync(
[EnumeratorCancellation] CancellationToken token)
{
using var dbConnection = new SqlConnection(_connectionString);
await dbConnection.OpenAsync(token);
 
using var command = dbConnection.CreateCommand();
command.CommandText = "SELECT * FROM HugeTable";
command.CommandTimeout = 300; // 5 минут
 
using var reader = await command.ExecuteReaderAsync(token);
 
// Отправляем записи по одной, а не собираем в коллекцию
while (await reader.ReadAsync(token))
{
    var item = new
    {
        Id = reader.GetInt32(0),
        Name = reader.GetString(1),
        Data = reader.GetString(2)
    };
    
    yield return JsonSerializer.Serialize(item);
}
}

private readonly JsonSerializerOptions _compactOptions = new()
{
DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
WriteIndented = false
};

public async IAsyncEnumerable<byte[]> StreamJsonRecordsAsync(
[EnumeratorCancellation] CancellationToken token)
{
foreach (var record in _dataService.GetRecords())
{
    using var memoryStream = new MemoryStream();
    using var writer = new Utf8JsonWriter(memoryStream);
    
    writer.WriteStartObject();
    writer.WriteNumber("id", record.Id);
    writer.WriteString("name", record.Name);
    writer.WriteEndObject();
    writer.Flush();
    
    yield return memoryStream.ToArray();
}
}

// На стороне клиента (JavaScript)
let lastProcessedId = 0;
 
eventSource.addEventListener('log', event => {
const logEntry = JSON.parse(event.data);
lastProcessedId = logEntry.id;
 
// Периодически отправляем подтверждение на сервер
if (logEntry.id % 100 === 0) {
    fetch('/api/logs/ack?lastId=' + lastProcessedId);
}
});

// Интеграция между SignalR и SSE
public class HybridCommunicationService
{
    private readonly IHubContext<CommandHub> _hubContext;
    private readonly Channel<string> _eventsChannel;
 
    public HybridCommunicationService(IHubContext<CommandHub> hubContext)
    {
        _hubContext = hubContext;
        _eventsChannel = Channel.CreateUnbounded<string>();
    }
 
    // Отправка команды через SignalR
    public async Task SendCommandAsync(string target, string command, object payload)
    {
        await _hubContext.Clients.User(target).SendAsync(command, payload);
    }
 
    // Публикация события для SSE
    public async Task PublishEventAsync(string eventData)
    {
        await _eventsChannel.Writer.WriteAsync(eventData);
    }
 
    // Подписка на события через SSE
    public IAsyncEnumerable<string> SubscribeToEvents(CancellationToken token)
    {
        return _eventsChannel.Reader.ReadAllAsync(token);
    }
}

public class ChatService
{
    private readonly List<ChatMessage> _messages = new();
    private readonly Channel<ChatMessage> _messageChannel = Channel.CreateUnbounded<ChatMessage>();
 
    public async Task AddMessageAsync(string sender, string text)
    {
        var message = new ChatMessage
        {
            Id = Guid.NewGuid(),
            Sender = sender,
            Text = text,
            Timestamp = DateTime.UtcNow
        };
        
        _messages.Add(message);
        await _messageChannel.Writer.WriteAsync(message);
    }
 
    public IAsyncEnumerable<string> GetMessagesStream(CancellationToken token)
    {
        return _messageChannel.Reader
            .ReadAllAsync(token)
            .Select(msg => JsonSerializer.Serialize(msg));
    }
    
    public List<ChatMessage> GetRecentMessages(int count = 10)
    {
        return _messages.OrderByDescending(m => m.Timestamp)
            .Take(count)
            .ToList();
    }
}

app.MapGet("/chat/messages", (ChatService chat) => 
    Results.Ok(chat.GetRecentMessages()));
 
app.MapPost("/chat/messages", async (ChatMessage message, ChatService chat) => 
{
    await chat.AddMessageAsync(message.Sender, message.Text);
    return Results.Ok();
});
 
app.MapGet("/chat/updates", (ChatService chat, CancellationToken token) =>
    TypedResults.ServerSentEvents(chat.GetMessagesStream(token), "chatMessage"));

SSE:            CPU: 15-20%, RAM: 1.2 GB
WebSocket:      CPU: 25-35%, RAM: 1.8 GB
Long-polling:   CPU: 40-55%, RAM: 2.3 GB

private readonly ConcurrentDictionary<string, (DateTime LastActivity, SseConnection Connection)> _connections = new();
 
public void CleanupInactiveConnections()
{
var threshold = DateTime.UtcNow.AddMinutes(-5);
var inactiveKeys = _connections.Where(kvp => kvp.Value.LastActivity < threshold)
                              .Select(kvp => kvp.Key).ToList();
 
foreach (var key in inactiveKeys)
{
    _connections.TryRemove(key, out _);
}
}

builder.Services.AddSingleton<ISseBackplane, RedisSseBackplane>();

foreach (var group in _subscribers.GroupBy(s => s.Topic))
{
var message = JsonSerializer.Serialize(new { Topic = group.Key, Data = data });
Parallel.ForEach(group, subscriber => subscriber.Channel.Writer.TryWrite(message));
}

public class SystemMetric
{
    public DateTime Timestamp { get; set; }
    public double CpuUsage { get; set; }
    public double MemoryUsage { get; set; }
    public int ActiveConnections { get; set; }
}
 
public class MetricsService
{
    private readonly Channel<SystemMetric> _metricsChannel = 
        Channel.CreateUnbounded<SystemMetric>();
 
    public async ValueTask PublishMetricAsync(SystemMetric metric)
    {
        await _metricsChannel.Writer.WriteAsync(metric);
    }
 
    public IAsyncEnumerable<string> GetMetricsStream(CancellationToken token)
    {
        return _metricsChannel.Reader
            .ReadAllAsync(token)
            .Select(m => JsonSerializer.Serialize(m));
    }
}
 
public class MetricsGenerator : BackgroundService
{
    private readonly MetricsService _metricsService;
    private readonly Random _random = new();
 
    public MetricsGenerator(MetricsService metricsService)
    {
        _metricsService = metricsService;
    }
 
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            var metric = new SystemMetric
            {
                Timestamp = DateTime.Now,
                CpuUsage = Math.Round(_random.NextDouble() * 100, 1),
                MemoryUsage = Math.Round(_random.NextDouble() * 16384, 0),
                ActiveConnections = _random.Next(10, 1000)
            };
 
            await _metricsService.PublishMetricAsync(metric);
            await Task.Delay(1000, stoppingToken);
        }
    }
}

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddSingleton<MetricsService>();
builder.Services.AddHostedService<MetricsGenerator>();
 
var app = builder.Build();
app.UseDefaultFiles();
app.UseStaticFiles();
 
app.MapGet("/metrics/stream", (MetricsService metrics, CancellationToken token) =>
    TypedResults.ServerSentEvents(
        metrics.GetMetricsStream(token), 
        eventType: "metric")
);
 
app.Run();