如何发送到 web api 控制器中的特定信号器客户端？

Question

我想向特定客户端发送数据。我有 Asp.net core web api(.Net-6.0) 控制器，它有一个集线器帮助调用远程 Worker 服务上的方法。 Hub 正在主动向特定的 Worker 客户端逐一发送调用。 如何以及在哪里保留 connectionId 和相应的 WorkerID，这样当 MiniAppController 收到请求时，它会使用 hubContext 通过正确的连接触发请求。代码示例是：

public class ChatHub : Hub
{
    private readonly ILogger<ChatHub> _logger;
    public ChatHub(ILogger<ChatHub> logger)
    {
        _logger = logger;
    }

    public async Task HandShake(string workerId, string message)
    {
        HubCallerContext context = this.Context;
        await Clients.Caller.SendAsync("HandShake", workerId, context.ConnectionId);
    }

    public override async Task OnConnectedAsync()
    {
        await Groups.AddToGroupAsync(Context.ConnectionId, "SignalR Users");

        await base.OnConnectedAsync();
    }
    public override async Task OnDisconnectedAsync(Exception exception)
    {
        await Groups.RemoveFromGroupAsync(Context.ConnectionId, "SignalR Users");
        _logger.LogInformation($"1.Server: Client disconnected  and left the group..............");
        await base.OnDisconnectedAsync(exception);
    }
} 

Webapi 控制器：

    [Route("api/[controller]")]
[ApiController]
public class MiniAppController : ControllerBase
{
    private readonly IHubContext<ChatHub> _chatHubContext;
    private readonly ILogger<ChatHub> _logger;

    public MiniAppController(IHubContext<ChatHub> chatHubContext)
    {
        _chatHubContext = chatHubContext;
    }
    [HttpGet]
    public async Task<ActionResult<CheckoutInfo>> Checkout(string comID, string parkServerID, string parkLotID, string parkID, string miniAppID, string miniUserID, string sign)
    {
        string workerId = comID + parkServerID + parkLotID;//extracted from the method arguments
        ***//how to use workerId to send to a specific client???***
        ......
    }
}

作为 SignalR 客户端的工作器服务，我可以有多个工作器：

    public class Worker1 : BackgroundService
{
    private readonly ILogger<Worker1> _logger;
    private HubConnection _connection;

    public Worker1(ILogger<Worker1> logger)
    {
        _logger = logger;

        _connection = new HubConnectionBuilder()
            .WithUrl("http://localhost:5106/chatHub")
            .WithAutomaticReconnect()
            .Build();

        _connection.On<string, string>("HandShakeAck", HandShakeAck);
        _connection.On<string, string>("ReceiveMessage", ReceiveMessage);
        _connection.On<CheckoutRequest>("Checkout", Checkout);
    }
    public Task Checkout(CheckoutRequest checkoutRequest)
    {
        //send Checkoutinfo back
        CheckoutInfo checkoutInfo = new CheckoutInfo();
        _connection.InvokeAsync("ReceiveCheckoutInfo", workerId, checkoutInfo);
        return Task.CompletedTask;
    }
}

请帮忙。谢谢

Answer 1

我认为最好的方法是跟踪 signalR 组中的连接。每次建立连接时，我们都需要按建立连接的 workerId 对其进行分组。最好在 onConnectedAsync 方法中执行此操作，因为这样我们就不必在每次重置连接时手动执行此操作。

但是我们如何知道在 onConnectedAsync 方法中连接的是哪个 worker？就像我在我的应用程序中使用访问令牌知道哪个用户正在连接一样。

不过要提一提的是，当使用此访问令牌时，SignalR 将在使用 websockets 连接时将其作为查询参数。如果您有 IIS 记录您的活动连接并且您认为工作人员 ID 敏感，您可能希望也可能不希望这样做。 （当使用长轮询或 SSE 时，访问令牌将在请求的标头中发送。

因此：

您可以在启动连接时将工作人员 ID 作为访问令牌传递。

    _connection = new HubConnectionBuilder()
        .WithUrl("http://localhost:5106/chatHub", options =>
         { 
            options.AccessTokenProvider = () => // pass worker id here;
         })
        .WithAutomaticReconnect()
        .Build();

注意：您可以选择加密工作人员 ID 并在服务器端解密。

如果您不想将 workerId 与访问令牌相关联，那么您也可以将其作为硬编码查询参数传递。 （这样做会将其保留为所有 3 种 signalR 连接类型的查询参数）。

    _connection = new HubConnectionBuilder()
        .WithUrl($"http://localhost:5106/chatHub?workerId={workerId}")
        .WithAutomaticReconnect()
        .Build();

您也可以使用完全成熟的 JWT 令牌，如果您愿意，也可以将 workerId 嵌入到 JWT 令牌中。

下一步是在 onConnectedAsync 方法中获取此工作人员 ID。为此，我们需要：

• workerId 中间件（获取workerId）
• workerId 服务（存储和访问workerId）
• workerId 要求属性（强制某些集线器方法存在workerId）
• WorkerId 中间件结果处理程序（如果 workerId 要求失败必须发生什么）

WorkerIdMiddleware 可以获取每个请求的worker id并将其存储在请求上下文中：

WorkerIdMiddleware.cs

public class WorkerIdMiddleware
{
    private readonly RequestDelegate _next;

    public WorkerIdMiddleware(RequestDelegate next)
    {
        _next = next;
    }

    public async Task Invoke(HttpContext httpContext)
    {
        var workerId = httpContext.Request.Query["access_token"];

        if (!string.IsNullOrEmpty(workerId))
        {
            AttachWorkerIdToContext(httpContext, workerId);
        }

        await _next(httpContext);
    }

    private void AttachWorkerIdToContext(HttpContext httpContext, string workerId)
    {
        if (ValidWorkerId(workerId))
        {
            httpContext.Items["WorkerId"] = workerId;
        }
    }

    private bool ValidWorkerId(string workerId)
    {
        // Validate the worker id if you need to
    }
}

然后我们可以通过 WorkerIdService 访问 workerId：

WorkerIdService.cs

public class WorkerIdService
{
    private string _currentWorkerId;
    private readonly IHttpContextAccessor _httpContextAccessor;

    public WorkerIdService(IHttpContextAccessor httpContextAccessor)
    {
        _httpContextAccessor = httpContextAccessor;
        _currentWorkerId = GetCurrentWorkerIdFromHttpContext();
    }

    public string CurrentWorkerId
    {
        get
        {
            if (_currentWorkerId == null)
            {
                _currentWorkerId = GetCurrentWorkerIdFromHttpContext();
            }

            return _currentWorkerId;
        }
    }


    private string GetCurrentWorkerIdFromHttpContext()
    {
        return (string)_httpContextAccessor.HttpContext?.Items?["WorkerId"];
    }
}

workerId 需求和需求处理程序将允许我们保护我们的 signalR 方法并确保在需要时传递一个 worker id：

ChatHubWorkerIdRequirement.cs

using Microsoft.AspNetCore.Authorization;

public class ChatHubWorkerIdRequirement : IAuthorizationRequirement
{
}

ChatHubWorkerIdHandler.cs

public class ChatHubWorkerIdHandler : AuthorizationHandler<ChatHubWorkerIdRequirement>
{
    readonly IHttpContextAccessor _httpContextAccessor;

    public ChatHubWorkerIdHandler(IHttpContextAccessor httpContextAccessor)
    {
        _httpContextAccessor = httpContextAccessor;
    }
    protected override Task HandleRequirementAsync(AuthorizationHandlerContext context, ChatHubWorkerIdRequirement requirement)
    {
        var workerId = (string)_httpContextAccessor.HttpContext.Items["WorkerId"];

        if (workerId != null)
        {
            // Connection may proceed successfully
            context.Succeed(requirement);
        }

        // Return completed task  
        return Task.CompletedTask;
    }
}

为了自定义workerId要求失败时响应的状态码，我们可以使用一个AuthorizationMiddlewareResultHandler

HubWorkerIdResponseHandler.cs

public class HubWorkerIdResponseHandler : IAuthorizationMiddlewareResultHandler
{
    private readonly IAuthorizationMiddlewareResultHandler _handler;

    public HubWorkerIdResponseHandler()
    {
        _handler = new AuthorizationMiddlewareResultHandler();
    }

    public async Task HandleAsync(
        RequestDelegate requestDelegate,
        HttpContext httpContext,
        AuthorizationPolicy authorizationPolicy,
        PolicyAuthorizationResult policyAuthorizationResult)
    {
        if (IsFailedPolicy(policyAuthorizationResult) && IsHubWorkerIdPolicy(authorizationPolicy))
        {
            // return whatever status code you wish if the hub is connected to without a worker id
            httpContext.Response.StatusCode = (int)HttpStatusCode.Unauthorized;
            return;
        }

        await _handler.HandleAsync(requestDelegate, httpContext, authorizationPolicy, policyAuthorizationResult);
    }

    private static bool IsFailedPolicy(PolicyAuthorizationResult policyAuthorizationResult)
    {
        return !policyAuthorizationResult.Succeeded;
    }

    private static bool IsHubWorkerIdPolicy(AuthorizationPolicy authorizationPolicy)
    {
        return authorizationPolicy.Requirements.OfType<ChatHubWorkerIdRequirement>().Any();
    }
}

最后，您需要像这样在您的启动中注册所有内容：

    public void ConfigureServices(IServiceCollection services)
    { 
        ...
        // Add the workerId policy
        services.AddSingleton<IAuthorizationHandler, ChatHubWorkerIdHandler>();
        services.AddAuthorization(options =>
        {
            options.AddPolicy("WorkerIdPolicy", policy =>
            {
                policy.Requirements.Add(new ChatHubWorkerIdRequirement());
            });
        });

        // Hub Policy failure response handler (this will handle the failed requirement above)
        services.AddSingleton<IAuthorizationMiddlewareResultHandler, HubWorkerIdResponseHandler>();

        services.AddSignalR();

        services.AddHttpContextAccessor();

        services.AddScoped<IWorkerIdService, WorkerIdService>();
    }

    public void Configure(IApplicationBuilder app)
    {
       ...
       app.UseMiddleware<JwtMiddleware>();
       app.UseAuthorization();
       app.UseEndpoints(endpoints =>
          ...
          endpoints.MapHub<ChatHub>("ChatHub"); 
      );
       ...
    }

您现在可以使用我们创建的新授权策略属性来装饰您的 ChatHub。通过装饰整个 hub 类，将在 onConnectedAsync 方法期间评估策略。

（如果您希望策略基于方法触发，则需要使用 workerId 策略属性装饰每个单独的方法）

[Authorize(Policy = "WorkerIdPolicy")]
public class ChatHub : Hub
{
    ....
}

然后您可以在 onConnectedAsync 方法期间从 WorkerIdService 访问 CurrentWorkerId：

public override async Task OnConnectedAsync()
{
    await Groups.AddToGroupAsync(Context.ConnectionId, "SignalR Users");
    // group the connections by workerId
    await Groups.AddToGroupAsync(Context.ConnectionId, $"Worker-{_workerIdService.CurrentWorkerId}");
    await base.OnConnectedAsync();
}

一切就绪后，您将能够使用 workerId 向该工作组发送信号，并且知道只有具有该 workerId 的客户端才能接收它。