如何在web控制器中发送到特定的signalr客户端？

Question

我想向特定的客户端发送数据。我有Asp.net核心web (.NET-6.0)控制器，它有一个集线器帮助调用远程工作人员服务上的方法。集线器是主动向特定的员工客户端逐个发送呼叫。如何以及在何处保存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方法中进行连接？通过使用访问令牌，我知道在我的应用程序中连接的是哪个用户。

但是，需要注意的一点是，当使用此访问令牌时，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 requirements属性)(强制对特定集线器methods)

• WorkerId中间件结果处理程序(如果workerId需求失败必须发生什么) workerId的存在)

WorkerIdMiddleware可以获取每个请求的辅助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
    }
}

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

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方法，并确保在需要时传递工作人员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。通过装饰整个集线器类，策略将在onConnectedAsync方法期间进行评估。

(如果希望在方法基础上触发策略，则需要使用workerId策略属性来修饰每个方法)

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

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

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的客户端/s才会接收到它。