Server-Sent Events (.NET 10)

using System.Net.ServerSentEvents;
using System.Runtime.CompilerServices;
using System.Threading.Channels;

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddSingleton<PriceTicker>();
var app = builder.Build();

app.MapGet("/products/{sku}/price-stream", (string sku, PriceTicker ticker, CancellationToken ct) =>
    TypedResults.ServerSentEvents(WithHeartbeat(ticker.Subscribe(sku, ct), TimeSpan.FromSeconds(20), ct)));

app.Run();

static async IAsyncEnumerable<SseItem<PriceChange?>> WithHeartbeat(
    ChannelReader<PriceChange> source, TimeSpan every, [EnumeratorCancellation] CancellationToken ct)
{
    while (!ct.IsCancellationRequested)
    {
        using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
        timeout.CancelAfter(every);
        PriceChange? next = null;
        try
        {
            if (!await source.WaitToReadAsync(timeout.Token)) yield break;
            source.TryRead(out next);
        }
        catch (OperationCanceledException) when (!ct.IsCancellationRequested)
        {
            // Nothing happened for 20 seconds: fall through and send a heartbeat.
        }

        yield return next is null
            ? new SseItem<PriceChange?>(null, "heartbeat")
            : new SseItem<PriceChange?>(next, "price");
    }
}

record PriceChange(string Sku, decimal Price, DateTimeOffset At);

sealed class PriceTicker
{
    public ChannelReader<PriceChange> Subscribe(string sku, CancellationToken ct)
    {
        // A real ticker registers this channel with the price source and unregisters it when ct fires.
        var channel = Channel.CreateBounded<PriceChange>(new BoundedChannelOptions(100) { FullMode = BoundedChannelFullMode.DropOldest });
        return channel.Reader;
    }
}

using System.Net.ServerSentEvents;
using System.Text.Json;

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddHttpClient("orders", c => c.BaseAddress = new Uri("https://orders.internal.example"));
builder.Services.AddHostedService<ShipmentListener>();
var app = builder.Build();
app.Run();

sealed class ShipmentListener(IHttpClientFactory factory, ILogger<ShipmentListener> logger) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var client = factory.CreateClient("orders");
        string? lastEventId = null;
        var failures = 0;

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                using var request = new HttpRequestMessage(HttpMethod.Get, "/orders/2042/events");
                if (lastEventId is not null) request.Headers.Add("Last-Event-ID", lastEventId);

                using var response = await client.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, stoppingToken);
                if (response.StatusCode == System.Net.HttpStatusCode.NoContent) return; // nothing more will come
                // A 401, 404 or 500 is not an empty stream: throw so it takes the backoff path.
                response.EnsureSuccessStatusCode();

                await using var stream = await response.Content.ReadAsStreamAsync(stoppingToken);
                var parser = SseParser.Create(stream, (type, data) => JsonSerializer.Deserialize<ShipmentEvent>(data, JsonSerializerOptions.Web));
                await foreach (var item in parser.EnumerateAsync(stoppingToken))
                {
                    lastEventId = item.EventId ?? lastEventId;
                    failures = 0;
                    logger.LogInformation("Order {OrderId} is now {Status}", item.Data?.OrderId, item.Data?.Status);
                }
                logger.LogInformation("Stream ended; reconnecting from event {LastEventId}", lastEventId);
            }
            // A mid-stream drop surfaces from the read as HttpIOException (an IOException), not
            // HttpRequestException; letting it escape would stop the whole host.
            catch (Exception ex) when (ex is HttpRequestException or IOException && !stoppingToken.IsCancellationRequested)
            {
                failures++;
                logger.LogWarning(ex, "Stream dropped; reconnecting from event {LastEventId}", lastEventId);
            }

            // Every path that loops waits first: exponential backoff capped at 60 s, plus jitter
            // so a fleet of listeners does not reconnect in lockstep after an upstream restart.
            var backoff = TimeSpan.FromSeconds(Math.Min(60, 3 * Math.Pow(2, Math.Min(failures, 5))));
            await Task.Delay(backoff + TimeSpan.FromMilliseconds(Random.Shared.Next(0, 1000)), stoppingToken);
        }
    }
}

record ShipmentEvent(int OrderId, string Status, DateTimeOffset At);

GET /orders/2042/events HTTP/1.1
Host: shop.example.com
Accept: text/event-stream
Cookie: .AspNetCore.Cookies=<session-cookie>
Last-Event-ID: 3

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-store, no-cache
X-Accel-Buffering: no

event: status
data: {"orderId":2042,"status":"OutForDelivery","at":"2026-10-22T09:14:03+00:00"}
id: 4
retry: 3000

event: status
data: {"orderId":2042,"status":"Delivered","at":"2026-10-22T11:40:51+00:00"}
id: 5
retry: 3000