SignalR 实时通信

// 服务端：注册 SignalR 与 Hub 端点
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddSignalR(options =>
{
    options.EnableDetailedErrors = builder.Environment.IsDevelopment();
    options.KeepAliveInterval = TimeSpan.FromSeconds(15);
    options.ClientTimeoutInterval = TimeSpan.FromSeconds(30);
    options.MaximumReceiveMessageSize = 32 * 1024;
});

var app = builder.Build();
app.MapHub<ChatHub>("/hubs/chat");
app.Run();

// 客户端：观察实际使用的传输
var connection = new HubConnectionBuilder()
    .WithUrl("https://localhost:5001/hubs/chat", options =>
    {
        options.Transports =
            HttpTransportType.WebSockets |
            HttpTransportType.ServerSentEvents |
            HttpTransportType.LongPolling;
    })
    .WithAutomaticReconnect()
    .Build();

connection.StartAsync().Wait();
Console.WriteLine($"传输方式: {connection.Transport?.GetType().Name}");

// 客户端契约
public interface IChatClient
{
    Task ReceiveMessage(string user, string message);
    Task UserJoined(string connectionId, string user);
    Task UserLeft(string connectionId, string user);
    Task ReceiveTyping(string user);
}

// 强类型 Hub
public class ChatHub : Hub<IChatClient>
{
    private readonly IChatService _service;

    public ChatHub(IChatService service) => _service = service;

    public override async Task OnConnectedAsync()
    {
        await Clients.Others.UserJoined(Context.ConnectionId, "匿名用户");
        await base.OnConnectedAsync();
    }

    public override async Task OnDisconnectedAsync(Exception? ex)
    {
        await Clients.Others.UserLeft(Context.ConnectionId, "匿名用户");
        await base.OnDisconnectedAsync(ex);
    }

    // 客户端可调用
    public async Task SendMessage(string user, string message)
    {
        await _service.SaveAsync(user, message);
        await Clients.All.ReceiveMessage(user, message);
    }

    public Task Typing() => Clients.Others.ReceiveTyping("匿名用户");
}

public class OrderHub : Hub<IOrderClient>
{
    public async Task SubscribeToOrder(string orderId)
    {
        // 加入订单专属组
        await Groups.AddToGroupAsync(Context.ConnectionId, $"order:{orderId}");
    }

    public async Task UnsubscribeFromOrder(string orderId)
    {
        await Groups.RemoveFromGroupAsync(Context.ConnectionId, $"order:{orderId}");
    }

    // 服务端主动推送（由后台任务调用）
    public static async Task BroadcastStatusAsync(
        IHubContext<OrderHub, IOrderClient> hub, string orderId, string status)
    {
        await hub.Clients.Group($"order:{orderId}").StatusChanged(orderId, status);
    }
}

// 用户定向：基于认证身份
public async Task NotifyUser(string userId, string message)
{
    await Clients.User(userId).ReceiveMessage("system", message);
}

// 多组/多用户组合
public async Task NotifyMany(string[] users, string message)
{
    await Clients.Users(users).ReceiveMessage("system", message);
}

// 客户端：完整的生命周期处理
var connection = new HubConnectionBuilder()
    .WithUrl("/hubs/chat")
    .WithAutomaticReconnect(new[] {
        TimeSpan.Zero,
        TimeSpan.FromSeconds(2),
        TimeSpan.FromSeconds(5),
        TimeSpan.FromSeconds(10)
    })
    .Build();

connection.Reconnecting += error =>
{
    Console.WriteLine($"重连中: {error?.Message}");
    return Task.CompletedTask;
};

connection.Reconnected += async connectionId =>
{
    Console.WriteLine($"已重连，新连接: {connectionId}");
    // 关键：重建订阅
    await connection.InvokeAsync("SubscribeToOrder", orderId);
};

connection.Closed += async error =>
{
    Console.WriteLine($"连接关闭: {error?.Message}");
    await Task.Delay(TimeSpan.FromSeconds(5));
    await connection.StartAsync();   // 手动兜底重连
};

await connection.StartAsync();

// 服务端流：持续推送行情
public async IAsyncEnumerable<Quote> StreamQuotes(
    string symbol,
    [EnumeratorCancellation] CancellationToken ct)
{
    while (!ct.IsCancellationRequested)
    {
        yield return await _market.GetQuoteAsync(symbol, ct);
        await Task.Delay(1000, ct);
    }
}

// 带背压的服务端流：Channel 作缓冲
public async IAsyncEnumerable<LogEntry> StreamLogs(
    [EnumeratorCancellation] CancellationToken ct)
{
    var channel = Channel.CreateBounded<LogEntry>(new BoundedChannelOptions(100)
    {
        FullMode = BoundedChannelFullMode.DropOldest
    });

    _ = _logSource.PumpIntoAsync(channel.Writer, ct);

    await foreach (var entry in channel.Reader.ReadAllAsync(ct))
    {
        yield return entry;
    }
}

// 客户端流：上传分片
public async Task UploadAsync(IAsyncEnumerable<Chunk> stream)
{
    await foreach (var chunk in stream)
    {
        await _storage.AppendAsync(chunk);
    }
}

// 接入 Redis 底板
builder.Services.AddSignalR()
    .AddStackExchangeRedis(options =>
    {
        options.Configuration.ChannelPrefix = RedisChannel.Literal("MyApp");
        options.Configuration.EndPoints.Add("redis:6379");
    });

// Azure SignalR Service：托管方案，无自建 Redis
// builder.Services.AddSignalR().AddAzureSignalR(connectionString);

// 多实例下服务端主动推送：通过 IHubContext
public class OrderStatusWorker : BackgroundService
{
    private readonly IHubContext<OrderHub, IOrderClient> _hub;
    private readonly IConnectionTracker _tracker;

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        await foreach (var evt in _bus.SubscribeAsync<OrderEvent>(ct))
        {
            // 底板会把消息路由到持有该组的实例
            await _hub.Clients.Group($"order:{evt.OrderId}")
                .StatusChanged(evt.OrderId, evt.Status);
        }
    }
}

// 诊断配置：打开详细日志与连接计数
builder.Services.AddSignalR(o => o.EnableDetailedErrors = true);

builder.Services.AddSingleton<IConnectionCounter, ConnectionCounter>();

app.MapHub<ChatHub>("/hubs/chat");

// 连接计数器：便于暴露成指标
public class ConnectionCounter : IConnectionCounter
{
    private int _current;

    public int Current => Volatile.Read(ref _current);
    public void OnConnected() => Interlocked.Increment(ref _current);
    public void OnDisconnected() => Interlocked.Decrement(ref _current);
}

{
  "Logging": {
    "LogLevel": {
      "Microsoft.AspNetCore.SignalR": "Debug",
      "Microsoft.AspNetCore.Http.Connections": "Debug"
    }
  }
}