diff --git a/BootstrapBlazor.Extensions.slnx b/BootstrapBlazor.Extensions.slnx index 088bbf7d..35ea650e 100644 --- a/BootstrapBlazor.Extensions.slnx +++ b/BootstrapBlazor.Extensions.slnx @@ -84,6 +84,7 @@ + diff --git a/src/extensions/BootstrapBlazor.SignalR.Client/BootstrapBlazor.SignalR.Client.csproj b/src/extensions/BootstrapBlazor.SignalR.Client/BootstrapBlazor.SignalR.Client.csproj new file mode 100644 index 00000000..511d95b1 --- /dev/null +++ b/src/extensions/BootstrapBlazor.SignalR.Client/BootstrapBlazor.SignalR.Client.csproj @@ -0,0 +1,39 @@ + + + + 11.0.0-rc.3.1 + + + + Optional SignalR client transport for BootstrapBlazor dispatch and connection services. + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/extensions/BootstrapBlazor.SignalR.Client/DefaultConnectionService.cs b/src/extensions/BootstrapBlazor.SignalR.Client/DefaultConnectionService.cs new file mode 100644 index 00000000..2fea2e53 --- /dev/null +++ b/src/extensions/BootstrapBlazor.SignalR.Client/DefaultConnectionService.cs @@ -0,0 +1,148 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the Apache 2.0 License +// See the LICENSE file in the project root for more information. +// Maintainer: Argo Zhang(argo@live.ca) Website: https://www.blazor.zone + +using Microsoft.Extensions.Logging; + +namespace BootstrapBlazor.Components; + +/// +/// 基于 SignalR 的客户端连接服务，通过心跳上报连接信息并轮询服务端连接快照 +/// SignalR client connection service that reports connection information through heartbeats and polls server connection snapshots +/// +sealed class DefaultConnectionService : IConnectionService, IAsyncDisposable +{ + private readonly SignalRConnection _connection; + private readonly ILogger _logger; + private readonly CancellationTokenSource _lifetime = new(); + private readonly Task _pollingTask; + private ConnectionSnapshot _snapshot = new(0, []); + private int _disposed; + + /// + /// 初始化客户端连接服务并启动连接快照轮询 + /// Initialize the client connection service and start connection snapshot polling + /// + /// 共享的 SignalR 连接The shared SignalR connection + /// 日志记录器The logger + public DefaultConnectionService(SignalRConnection connection, ILogger logger) + { + _connection = connection; + _logger = logger; + _connection.Retain(); + _pollingTask = PollAsync(_lifetime.Token); + } + + /// + /// 获得 最近一次服务端快照中尚未超时的在线连接数量 + /// Gets the online connection count from the latest server snapshot, excluding timed-out connections + /// + public long Count => _snapshot.Count; + + /// + /// 获得 最近一次服务端查询的连接集合 + /// Gets the connection collection from the latest server query + /// + public ICollection Connections => _snapshot.Connections; + + /// + /// 根据连接标识从本地快照中获取连接信息 + /// Try to get connection information from the local snapshot by connection identifier + /// + /// 连接标识The connection identifier + /// 匹配的连接信息，未找到时为 nullThe matching connection information, or null if not found + /// 找到连接时返回 true，否则返回 falseTrue if the connection is found; otherwise, false + public bool TryGetValue(string key, [NotNullWhen(true)] out ConnectionItem? value) + { + value = _snapshot.Connections.FirstOrDefault(item => item.Id == key); + return value != null; + } + + /// + /// 向服务端发送心跳以添加或更新客户端连接信息 + /// Send a heartbeat to the server to add or update client connection information + /// + /// 客户端信息The client information + /// 取消令牌The cancellation token + /// 表示异步操作的任务A task representing the asynchronous operation + public async Task AddOrUpdateAsync(ClientInfo client, CancellationToken cancellationToken = default) + { + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _lifetime.Token); + if (!await _connection.InvokeAsync("Heartbeat", client, cancellation.Token)) + { + _logger.LogWarning("Connection heartbeats are disabled on the server."); + } + } + + /// + /// 定期刷新连接快照，查询被禁用、未获授权或请求被拒绝时停止轮询 + /// Periodically refresh the connection snapshot, stopping when queries are disabled, unauthorized, or rejected + /// + /// 取消令牌The cancellation token + /// 表示轮询操作的任务A task representing the polling operation + private async Task PollAsync(CancellationToken cancellationToken) + { + try + { + while (!cancellationToken.IsCancellationRequested) + { + try + { + var snapshot = await _connection.InvokeAsync("GetConnectionSnapshot", null, cancellationToken); + if (snapshot == null) + { + _snapshot = new(0, []); + _logger.LogInformation("Connection snapshot polling stopped because queries are disabled or not authorized."); + return; + } + _snapshot = snapshot; + } + catch (HttpRequestException ex) when (!cancellationToken.IsCancellationRequested && !SignalRConnection.IsTransientFailure(ex)) + { + _snapshot = new(0, []); + _logger.LogWarning(ex, "Connection snapshot polling stopped because the Hub request was rejected."); + return; + } + catch (Exception ex) when (!cancellationToken.IsCancellationRequested) + { + _logger.LogError(ex, "Unable to refresh the connection snapshot."); + } + await Task.Delay(TimeSpan.FromSeconds(3), cancellationToken); + } + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + } + } + + /// + /// 停止连接快照轮询并释放共享连接的使用引用 + /// Stop connection snapshot polling and release the usage reference to the shared connection + /// + /// 表示异步释放操作的任务A task representing the asynchronous disposal operation + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + { + return; + } + + _lifetime.Cancel(); + try + { + await _pollingTask; + } + finally + { + try + { + await _connection.ReleaseAsync(); + } + finally + { + _lifetime.Dispose(); + } + } + } +} diff --git a/src/extensions/BootstrapBlazor.SignalR.Client/DefaultDispatchService.cs b/src/extensions/BootstrapBlazor.SignalR.Client/DefaultDispatchService.cs new file mode 100644 index 00000000..3ff5563f --- /dev/null +++ b/src/extensions/BootstrapBlazor.SignalR.Client/DefaultDispatchService.cs @@ -0,0 +1,248 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the Apache 2.0 License +// See the LICENSE file in the project root for more information. +// Maintainer: Argo Zhang(argo@live.ca) Website: https://www.blazor.zone + +using Microsoft.Extensions.Logging; +using System.Text.Json; + +namespace BootstrapBlazor.Components; + +/// +/// 基于 SignalR 的客户端消息分发服务 +/// SignalR client dispatch service +/// +/// 消息数据类型The entry data type +/// 共享的 SignalR 连接The shared SignalR connection +/// 日志记录器The logger +sealed class DefaultDispatchService( + SignalRConnection connection, + ILogger> logger) : IDispatchService, IAsyncDisposable +{ + private static readonly string _entryType = typeof(TEntry).FullName ?? typeof(TEntry).Name; + private static readonly JsonSerializerOptions _serializerOptions = new(JsonSerializerDefaults.Web); + private readonly List _callbacks = []; + private readonly object _locker = new(); + private readonly SemaphoreSlim _subscriptions = new(1, 1); + private readonly CancellationTokenSource _lifetime = new(); + private bool _requiresUnsubscribe; + + /// + /// 向服务端分发消息，成功后通知本地订阅者 + /// Dispatch an entry to the server and notify local subscribers after a successful send + /// + /// 消息分发项The dispatch entry + /// 取消令牌The cancellation token + /// 表示异步操作的任务A task representing the asynchronous operation + public async Task DispatchAsync(DispatchEntry entry, CancellationToken cancellationToken = default) + { + await connection.DispatchAsync(_entryType, entry, cancellationToken); + await DispatchCoreAsync(entry, cancellationToken); + } + + /// + /// 注册消息回调，并在添加首个本地订阅者时订阅服务端消息 + /// Register a message callback and subscribe to server messages when adding the first local subscriber + /// + /// 消息回调委托The message callback + /// 取消令牌The cancellation token + /// 表示异步操作的任务A task representing the asynchronous operation + public async Task SubscribeAsync(Func, Task> callback, CancellationToken cancellationToken = default) + { + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _lifetime.Token); + var token = cancellation.Token; + // 串行处理订阅与退订，避免后续订阅者在首个远程订阅尚未就绪时返回成功。 + // Serialize subscription changes so later subscribers cannot succeed before the first remote subscription is ready. + await _subscriptions.WaitAsync(token); + try + { + var subscription = new Subscription(callback); + // first 表示本次添加前没有本地订阅者，而非服务生命周期内第一次订阅。 + // 同一消息类型仅在本地订阅数从 0 变为 1 时请求远程订阅，后续回调共用该订阅。 + // first means there were no local subscribers before this addition, not the first subscription in the service lifetime. + // Request a remote subscription only when the local count changes from 0 to 1; later callbacks share it. + bool first; + lock (_locker) + { + first = _callbacks.Count == 0; + _callbacks.Add(subscription); + } + if (first) + { + try + { + while (true) + { + // 这里不能写 while(!token.IsCancellationRequested) 因为取消时仍需清理远程订阅。 + // Do not write while(!token.IsCancellationRequested) here because remote subscription still needs to be cleaned up when canceled. + token.ThrowIfCancellationRequested(); + try + { + // 远程调用失败或取消时，服务端仍可能已完成订阅，因此提前标记需要清理。 + // The server may have subscribed even if the call fails or is canceled, so mark cleanup as required beforehand. + _requiresUnsubscribe = true; + await connection.SubscribeAsync(_entryType, ReceiveAsync, token); + break; + } + catch (Exception ex) when (!token.IsCancellationRequested && + (SignalRConnection.IsTransientFailure(ex) || + ex is InvalidOperationException && connection.IsReconnecting)) + { + logger.LogWarning(ex, "Unable to subscribe to dispatch entry type {EntryType}; retrying.", _entryType); + await Task.Delay(TimeSpan.FromSeconds(3), token); + } + } + } + catch + { + // 订阅未就绪时撤销本次本地注册，使后续订阅者仍能走 first 分支重新尝试。 + // Roll back this registration if subscribing fails so the next subscriber can retry through the first branch. + lock (_locker) + { + _callbacks.Remove(subscription); + } + throw; + } + } + } + finally + { + _subscriptions.Release(); + } + } + + /// + /// 移除消息回调，并在移除最后一个本地订阅者时取消服务端订阅 + /// Remove a message callback and unsubscribe from the server when removing the last local subscriber + /// + /// 消息回调委托The message callback + /// 取消令牌The cancellation token + /// 表示异步操作的任务A task representing the asynchronous operation + /// + /// 传输层退订取消或失败时保留待清理状态，再次退订或释放服务时继续清理 + /// If transport unsubscription is canceled or fails, retain the pending cleanup state for a subsequent unsubscription or service disposal + /// + public async Task UnSubscribeAsync(Func, Task> callback, CancellationToken cancellationToken = default) + { + await _subscriptions.WaitAsync(cancellationToken); + try + { + bool empty; + lock (_locker) + { + var index = _callbacks.FindIndex(subscription => subscription.Callback == callback); + if (index >= 0) + { + _callbacks.RemoveAt(index); + } + + // 集合中仍可能存在相同回调的其他订阅记录，因此仅在集合为空时才退订服务端。 + // There may still be other subscription records with the same callback in the collection, so unsubscribe from the server only when the collection is empty. + empty = _callbacks.Count == 0; + } + if (empty && _requiresUnsubscribe) + { + await connection.UnsubscribeAsync(_entryType, cancellationToken); + _requiresUnsubscribe = false; + } + } + finally + { + _subscriptions.Release(); + } + } + + /// + /// 反序列化服务端消息并通知本地订阅者 + /// Deserialize a server message and notify local subscribers + /// + /// 消息的 JSON 数据The message JSON data + /// 表示异步操作的任务A task representing the asynchronous operation + private Task ReceiveAsync(JsonElement json) + => DispatchCoreAsync(json.Deserialize>(_serializerOptions) + ?? throw new JsonException("The dispatch entry must not be null.")); + + /// + /// 依次调用本地消息回调，记录单个回调的异常并继续通知其他订阅者 + /// Invoke local message callbacks in order, logging individual callback failures and continuing to notify other subscribers + /// + /// 消息分发项The dispatch entry + /// 取消令牌The cancellation token + /// 表示异步操作的任务A task representing the asynchronous operation + private async Task DispatchCoreAsync(DispatchEntry entry, CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + + Subscription[] callbacks; + lock (_locker) + { + callbacks = _callbacks.ToArray(); + } + foreach (var subscription in callbacks) + { + cancellationToken.ThrowIfCancellationRequested(); + try + { + Task callbackTask; + lock (_locker) + { + cancellationToken.ThrowIfCancellationRequested(); + if (!_callbacks.Contains(subscription)) + { + continue; + } + + callbackTask = subscription.Callback(entry); + } + + await callbackTask.WaitAsync(cancellationToken); + cancellationToken.ThrowIfCancellationRequested(); + } + catch (Exception ex) when (!cancellationToken.IsCancellationRequested) + { + logger.LogError(ex, "A dispatch subscriber failed for entry type {EntryType}.", _entryType); + } + } + } + + /// + /// 取消正在进行的订阅操作，清空本地回调并清理传输层订阅，包括之前未完成的退订 + /// Cancel pending subscription operations, clear local callbacks, and clean up transport subscriptions, including previously incomplete unsubscriptions + /// + /// 表示异步释放操作的任务A task representing the asynchronous disposal operation + public async ValueTask DisposeAsync() + { + _lifetime.Cancel(); + await _subscriptions.WaitAsync(); + try + { + lock (_locker) + { + _callbacks.Clear(); + } + if (_requiresUnsubscribe) + { + await connection.UnsubscribeAsync(_entryType, CancellationToken.None); + _requiresUnsubscribe = false; + } + } + finally + { + _subscriptions.Release(); + } + } + + /// + /// 独立的消息订阅记录，用于区分同一回调的多次订阅 + /// An individual message subscription that distinguishes separate registrations of the same callback + /// + /// 消息回调委托The message callback + private sealed class Subscription(Func, Task> callback) + { + /// + /// 获得 消息回调委托 + /// Gets the message callback + /// + public Func, Task> Callback { get; } = callback; + } +} diff --git a/src/extensions/BootstrapBlazor.SignalR.Client/ServiceCollectionExtensions.cs b/src/extensions/BootstrapBlazor.SignalR.Client/ServiceCollectionExtensions.cs new file mode 100644 index 00000000..4c168530 --- /dev/null +++ b/src/extensions/BootstrapBlazor.SignalR.Client/ServiceCollectionExtensions.cs @@ -0,0 +1,39 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the Apache 2.0 License +// See the LICENSE file in the project root for more information. +// Maintainer: Argo Zhang(argo@live.ca) Website: https://www.blazor.zone + +using BootstrapBlazor.Components; +using Microsoft.Extensions.DependencyInjection.Extensions; + +namespace Microsoft.Extensions.DependencyInjection; + +/// +/// SignalR 客户端服务扩展类 +/// SignalR client service collection extensions +/// +public static class ServiceCollectionExtensions +{ + /// + /// 注册 SignalR 客户端消息分发、心跳上报与连接快照服务，共用一个 Hub 连接 + /// Register SignalR client dispatch, heartbeat reporting, and connection snapshot services with one shared Hub connection + /// + /// 服务集合The service collection + /// 服务集合The service collection + /// + /// 在 WebAssembly 宿主中调用 AddBootstrapBlazor 后调用此方法。离线发送会失败，消息不会排队或重放 + /// Call this method after AddBootstrapBlazor in the WebAssembly host. Offline sends fail; messages are not queued or replayed + /// 服务端需注册并映射对应 Hub，Subscribe 方法需返回 bool，以区分订阅拒绝与可重试的执行失败。连接快照查询需在服务端单独启用并授权客户端访问 + /// The server must register and map a corresponding Hub whose Subscribe method returns bool to distinguish subscription rejection from retryable execution failures. Connection snapshot queries require explicit server-side opt-in and client authorization + /// + public static IServiceCollection AddBootstrapBlazorHub(this IServiceCollection services) + { + // 增加 Connection 服务 + services.TryAddSingleton(); + services.AddSingleton(); + + // 增加 Dispatch 服务 + services.AddSingleton(typeof(IDispatchService<>), typeof(DefaultDispatchService<>)); + return services; + } +} diff --git a/src/extensions/BootstrapBlazor.SignalR.Client/SignalRConnection.cs b/src/extensions/BootstrapBlazor.SignalR.Client/SignalRConnection.cs new file mode 100644 index 00000000..2cb52b9c --- /dev/null +++ b/src/extensions/BootstrapBlazor.SignalR.Client/SignalRConnection.cs @@ -0,0 +1,396 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the Apache 2.0 License +// See the LICENSE file in the project root for more information. +// Maintainer: Argo Zhang(argo@live.ca) Website: https://www.blazor.zone + +using Microsoft.AspNetCore.Components; +using Microsoft.AspNetCore.SignalR; +using Microsoft.AspNetCore.SignalR.Client; +using Microsoft.Extensions.Logging; +using System.Collections.Concurrent; +using System.Net; +using System.Net.WebSockets; +using System.Text.Json; + +namespace BootstrapBlazor.Components; + +/// +/// 消息分发与连接服务共用的 SignalR Hub 连接 +/// SignalR Hub connection shared by dispatch and connection services +/// +sealed class SignalRConnection : IAsyncDisposable +{ + private readonly HubConnection _connection; + private readonly ILogger _logger; + private readonly ConcurrentDictionary> _subscriptions = new(StringComparer.Ordinal); + private readonly HashSet _pendingSubscriptions = new(StringComparer.Ordinal); + private readonly SemaphoreSlim _gate = new(1, 1); + private readonly CancellationTokenSource _lifetime = new(); + private readonly IDisposable _receiver; + private int _connectionUsers; + + /// + /// 获得 当前连接是否正在重连 + /// Gets whether the connection is reconnecting + /// + public bool IsReconnecting => _connection.State == HubConnectionState.Reconnecting; + + /// + /// 判断异常是否属于可重试的暂时性连接故障 + /// Determine whether an exception represents a transient connection failure that can be retried + /// + /// 待判断的异常The exception to check + /// 暂时性故障返回 true，否则返回 falseTrue for a transient failure; otherwise, false + internal static bool IsTransientFailure(Exception exception) => exception switch + { + HttpRequestException ex => ex.StatusCode is null or HttpStatusCode.RequestTimeout or + HttpStatusCode.TooManyRequests or HttpStatusCode.InternalServerError or + HttpStatusCode.BadGateway or HttpStatusCode.ServiceUnavailable or HttpStatusCode.GatewayTimeout, + IOException or WebSocketException or TimeoutException or OperationCanceledException => true, + _ => false + }; + + /// + /// 初始化共享连接，配置自动重连与消息接收回调 + /// Initialize the shared connection with automatic reconnection and a message receiver + /// + /// 导航管理器，用于解析 Hub 的绝对地址The navigation manager used to resolve the absolute Hub URL + /// 日志记录器The logger + public SignalRConnection(NavigationManager navigation, ILogger logger) + { + _logger = logger; + _connection = new HubConnectionBuilder() + .WithUrl(navigation.ToAbsoluteUri("bb.hub")) + .WithAutomaticReconnect(new ConnectionRetryPolicy()) + .Build(); + _receiver = _connection.On("Receive", ReceiveAsync); + _connection.Reconnected += RestoreSubscriptionsAsync; + } + + /// + /// 注册指定消息类型的接收回调并订阅服务端消息 + /// Register a receiver for the specified entry type and subscribe to server messages + /// + /// 消息类型标识The entry type identifier + /// 消息接收回调委托The message receiver callback + /// 取消令牌The cancellation token + /// 表示异步操作的任务A task representing the asynchronous operation + public async Task SubscribeAsync(string entryType, Func callback, CancellationToken cancellationToken) + { + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _lifetime.Token); + var token = cancellation.Token; + await _gate.WaitAsync(token); + try + { + _subscriptions[entryType] = callback; + await EnsureConnectedAsync(token); + await SubscribeCoreAsync(entryType, token); + _pendingSubscriptions.Remove(entryType); + } + catch + { + _subscriptions.TryRemove(entryType, out _); + _pendingSubscriptions.Remove(entryType); + throw; + } + finally + { + _gate.Release(); + } + } + + /// + /// 取消指定消息类型的订阅，并在无订阅和使用引用时停止连接 + /// Unsubscribe from the specified entry type and stop the connection when no subscriptions or usage references remain + /// + /// 消息类型标识The entry type identifier + /// 取消令牌The cancellation token + /// 表示异步操作的任务A task representing the asynchronous operation + public async Task UnsubscribeAsync(string entryType, CancellationToken cancellationToken) + { + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _lifetime.Token); + var token = cancellation.Token; + await _gate.WaitAsync(token); + try + { + _subscriptions.TryRemove(entryType, out _); + _pendingSubscriptions.Remove(entryType); + if (_connection.State == HubConnectionState.Connected) + { + await _connection.InvokeAsync("Unsubscribe", entryType, token); + } + if (_subscriptions.IsEmpty && Volatile.Read(ref _connectionUsers) == 0) + { + await _connection.StopAsync(token); + } + } + finally + { + _gate.Release(); + } + } + + /// + /// 增加共享连接的使用引用，防止取消消息订阅时停止仍在使用的连接 + /// Add a usage reference to prevent unsubscription from stopping a connection still in use + /// + public void Retain() => Interlocked.Increment(ref _connectionUsers); + + /// + /// 释放共享连接的使用引用，并在无使用引用和订阅时停止连接 + /// Release a usage reference and stop the connection when no usage references or subscriptions remain + /// + /// 表示异步操作的任务A task representing the asynchronous operation + public async Task ReleaseAsync() + { + await _gate.WaitAsync(_lifetime.Token); + try + { + if (Interlocked.Decrement(ref _connectionUsers) == 0 && _subscriptions.IsEmpty) + { + await _connection.StopAsync(_lifetime.Token); + } + } + finally + { + _gate.Release(); + } + } + + /// + /// 调用服务端 Hub 方法并返回结果，操作超时时间为 30 秒 + /// Invoke a server Hub method and return its result with a 30-second operation timeout + /// + /// 返回结果类型The result type + /// Hub 方法名称The Hub method name + /// 方法参数，为 null 时不传递参数The method argument; null invokes the method without arguments + /// 取消令牌The cancellation token + /// 服务端返回的结果The result returned by the server + public async Task InvokeAsync(string method, object? argument, CancellationToken cancellationToken) + { + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _lifetime.Token); + cancellation.CancelAfter(TimeSpan.FromSeconds(30)); + var token = cancellation.Token; + await _gate.WaitAsync(token); + try + { + await EnsureConnectedAsync(token); + return await _connection.InvokeCoreAsync(method, argument == null ? [] : [argument], token); + } + finally + { + _gate.Release(); + } + } + + /// + /// 通过共享 Hub 连接向服务端分发消息 + /// Dispatch an entry to the server through the shared Hub connection + /// + /// 消息数据类型The entry data type + /// 消息类型标识The entry type identifier + /// 消息分发项The dispatch entry + /// 取消令牌The cancellation token + /// 表示异步操作的任务A task representing the asynchronous operation + public async Task DispatchAsync(string entryType, DispatchEntry entry, CancellationToken cancellationToken) + { + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _lifetime.Token); + var token = cancellation.Token; + await _gate.WaitAsync(token); + try + { + await EnsureConnectedAsync(token); + await _connection.InvokeAsync("Dispatch", entryType, entry, token); + } + finally + { + _gate.Release(); + } + } + + /// + /// 在连接断开时尝试建立连接并恢复订阅，连接尚未就绪时抛出异常 + /// Attempt to start a disconnected connection and restore subscriptions, throwing if the connection is not ready + /// + /// 取消令牌The cancellation token + /// 表示异步操作的任务A task representing the asynchronous operation + private async Task EnsureConnectedAsync(CancellationToken cancellationToken) + { + if (_connection.State == HubConnectionState.Disconnected) + { + for (var attempt = 0; ; attempt++) + { + try + { + await _connection.StartAsync(cancellationToken); + break; + } + catch (Exception ex) when (attempt < 2 && !cancellationToken.IsCancellationRequested && IsTransientFailure(ex)) + { + _logger.LogWarning(ex, "Unable to connect to the BootstrapBlazor Hub; retrying."); + await Task.Delay(TimeSpan.FromSeconds(3), cancellationToken); + } + } + + _pendingSubscriptions.UnionWith(_subscriptions.Keys); + } + + if (_connection.State != HubConnectionState.Connected) + { + throw new InvalidOperationException("The BootstrapBlazor connection is reconnecting. Retry when it is connected."); + } + + await RestoreSubscriptionsCoreAsync(cancellationToken); + } + + /// + /// 自动重连成功后逐项恢复消息订阅，并重试暂时性故障 + /// Restore message subscriptions individually after automatic reconnection and retry transient failures + /// + /// 重连后的连接标识The connection identifier after reconnection + /// 表示异步操作的任务A task representing the asynchronous operation + private async Task RestoreSubscriptionsAsync(string? connectionId) + { + var token = _lifetime.Token; + try + { + var initialize = true; + while (true) + { + await _gate.WaitAsync(token); + try + { + if (initialize) + { + _pendingSubscriptions.UnionWith(_subscriptions.Keys); + initialize = false; + } + if (_connection.State != HubConnectionState.Connected) + { + return; + } + + await RestoreSubscriptionsCoreAsync(token); + if (_pendingSubscriptions.Count == 0) + { + return; + } + } + finally + { + _gate.Release(); + } + + await Task.Delay(TimeSpan.FromSeconds(3), token); + } + } + catch (OperationCanceledException) when (token.IsCancellationRequested) + { + } + catch (Exception ex) + { + _logger.LogError(ex, "Unable to restore dispatch subscriptions after reconnecting."); + } + } + + /// + /// 在持有连接锁时恢复待处理订阅，隔离单项失败；服务端明确拒绝时停止重试，执行失败时保留待重试项 + /// Restore pending subscriptions while holding the connection gate, isolating individual failures; stop retrying explicit server rejections and retain server execution failures for retry + /// + /// 取消令牌The cancellation token + /// 表示异步操作的任务A task representing the asynchronous operation + private async Task RestoreSubscriptionsCoreAsync(CancellationToken cancellationToken) + { + foreach (var entryType in _pendingSubscriptions.ToArray()) + { + cancellationToken.ThrowIfCancellationRequested(); + try + { + await SubscribeCoreAsync(entryType, cancellationToken); + _pendingSubscriptions.Remove(entryType); + } + catch (Exception ex) when (!cancellationToken.IsCancellationRequested) + { + if (!IsTransientFailure(ex) && _connection.State == HubConnectionState.Connected) + { + _pendingSubscriptions.Remove(entryType); + } + _logger.LogError(ex, "Unable to restore the dispatch subscription for entry type {EntryType}.", entryType); + } + } + } + + /// + /// 请求服务端订阅，区分明确拒绝与可重试的服务端执行失败 + /// Request a server subscription, distinguishing explicit rejection from retryable server execution failures + /// + /// 消息类型标识The entry type identifier + /// 取消令牌The cancellation token + /// 表示异步操作的任务A task representing the asynchronous operation + private async Task SubscribeCoreAsync(string entryType, CancellationToken cancellationToken) + { + try + { + if (!await _connection.InvokeAsync("Subscribe", entryType, cancellationToken)) + { + throw new UnauthorizedAccessException("Dispatch subscriptions are not allowed for this type."); + } + } + catch (HubException ex) + { + throw new IOException("The server could not complete the dispatch subscription.", ex); + } + } + + /// + /// 将服务端消息传递给对应消息类型的接收回调 + /// Forward a server message to the receiver registered for its entry type + /// + /// 消息类型标识The entry type identifier + /// 消息的 JSON 数据The message JSON data + /// 表示异步操作的任务A task representing the asynchronous operation + private async Task ReceiveAsync(string entryType, JsonElement entry) + { + if (_subscriptions.TryGetValue(entryType, out var callback)) + { + try + { + await callback(entry); + } + catch (Exception ex) + { + _logger.LogError(ex, "Unable to receive a dispatch for entry type {EntryType}.", entryType); + } + } + } + + /// + /// 取消连接操作，注销消息接收回调并释放 Hub 连接 + /// Cancel connection operations, unregister the message receiver, and dispose of the Hub connection + /// + /// 表示异步释放操作的任务A task representing the asynchronous disposal operation + public async ValueTask DisposeAsync() + { + _lifetime.Cancel(); + _connection.Reconnected -= RestoreSubscriptionsAsync; + _receiver.Dispose(); + await _connection.DisposeAsync(); + _lifetime.Dispose(); + } + + /// + /// 自动重连策略，首次立即重试，后续间隔递增至最多 30 秒 + /// Automatic reconnection policy that retries immediately at first, then increases the delay up to 30 seconds + /// + private sealed class ConnectionRetryPolicy : IRetryPolicy + { + /// + /// 根据已重试次数计算下次重连的等待时间 + /// Calculate the delay before the next reconnection attempt based on the previous retry count + /// + /// 重试上下文The retry context + /// 下次重连前的等待时间The delay before the next reconnection attempt + public TimeSpan? NextRetryDelay(RetryContext retryContext) + => TimeSpan.FromSeconds(Math.Min(30, retryContext.PreviousRetryCount * 3)); + } +} diff --git a/src/middleware/BootstrapBlazor.Middleware/ConnectionSnapshot.cs b/src/middleware/BootstrapBlazor.Middleware/ConnectionSnapshot.cs new file mode 100644 index 00000000..bb587709 --- /dev/null +++ b/src/middleware/BootstrapBlazor.Middleware/ConnectionSnapshot.cs @@ -0,0 +1,10 @@ +// Copyright (c) BootstrapBlazor & Argo Zhang (argo@live.ca). All rights reserved. +// Licensed under the Apache License, Version 2.0. + +namespace BootstrapBlazor.Components; + +/// +/// SignalR 连接快照及服务端计算的在线连接数量 +/// SignalR connection snapshot and the online connection count computed by the server +/// +public sealed record ConnectionSnapshot(long Count, ICollection Connections);