c# - 如何在 Controller 以外的地方调用 SignalR Clients.All.InvokeAsync()？

Question

我可以访问我的 IHubContext<MyHub>通过构造函数中的 DI 在我的 .NET Core WebAPI Controller 中很好，但我也想在其他地方访问它。

具体来说，当我使用来自 RabbitMQ 的消息时，有时我想通过 _myHubContext.Clients.All.InvokeAsync() 更新客户端，但我只是不知道如何获得它。

我在查找在 Controller 之外执行此类操作的文档时也遇到了问题。

如有任何帮助，我们将不胜感激。

编辑:

要添加一些细节，以及我的问题可能源自何处，我正在尝试访问我的 ConfigureServices 中的 IHubContext(以及我自己在 Startup 中注册的一些服务)类，特别是在 IApplicationLifetime 期间ApplicationStarted和 ApplicationStopped它调用 RabbitMQ 消费者的方法来连接和断开连接。

我的猜测是正确的，也许我无法访问 Startup 中的注册服务。类(class)？如果是这样，我将如何启动这些服务？

更新:

移动services.AddSignalR()以及一些在启动时调用的服务，直到 WebHost.ConfigureServices。在Program.cs解决了我的一些问题，当然还有更多。

当我收到来自 RabbitMQ 的消息时，我的 JS 客户端没有收到任何消息，但我的客户端已成功连接。 “奇怪……”我想。为了获得更多信息，我在我的 Controller 中连接了一个 GET 操作以通过 SignalR Hub 发送一些内容。 .每当我调用 GET 时，它都会工作...... IHubContext<MyHub> .我得到 hubContext通过我的 RabbitMQ 监听器中的构造函数，就像我对 Controller 所做的那样。

新问题: Controller 中注入(inject)的东西与我在启动时自己注册的服务中注入(inject)的东西是否不同？怎么会这样，我该如何克服？

一些代码......

摘自 Program.cs

public static IWebHost BuildWebHost(string[] args) =>
    WebHost.CreateDefaultBuilder(args)
        .UseKestrel()
        .UseIISIntegration()
        .ConfigureServices(services => {
            services.AddSignalR();
            services.AddTransient<ISubscriber, Subscriber>();
            services.AddTransient<IDataService, DataService>();
            services.AddTransient<IHealthCheckProcessor, HealthCheckProcessor>();
            services.AddTransient<INodeProcessor, NodeProcessor>();
        })
        .UseStartup<Startup>()
        .Build();

来自 Startup.cs

public class Startup
{
    public Startup(IConfiguration _configuration, ISubscriber _subscriber)
    {
        configuration = _configuration;
        subscriber = _subscriber;
    }
    public IConfiguration configuration { get; }
    public ISubscriber subscriber { get; }

    public void ConfigureServices(IServiceCollection services)
    {
        services.AddCors();
        services.AddMvc();
    }

    public void Configure(IApplicationBuilder app, IHostingEnvironment env, IApplicationLifetime applicationLifetime)
    {
        if (env.IsDevelopment())
        {
            app.UseDeveloperExceptionPage();
        }
        app.UseCors(builder => builder
            // CORS stuff);

        app.UseSignalR(routes =>
        {
            routes.MapHub<StatusHub>("Status");
        });
        app.UseMvc();
        applicationLifetime.ApplicationStarted.Register(OnStartup);
        applicationLifetime.ApplicationStopping.Register(OnShutdown);
    }

    private void OnStartup() {
        // MessageBroker stuff
        subscriber.Start(messageBroker);
    }

    private void OnShutdown() {
        subscriber.Stop();
    }
}

来自 Subscriber.cs

public class Subscriber : ISubscriber {
    public static IConnection connection;
    public static IModel channel;
    public IHubContext<StatusHub> hubContext;

    public static IHealthCheckProcessor healthCheckProcessor;
    public static INodeProcessor nodeProcessor;

    public Subscriber(IHubContext<StatusHub> _hubContext, INodeProcessor _nodeProcessor, IHealthCheckProcessor _healthCheckProcessor)
    {
        connection = new ConnectionFactory().CreateConnection();
        channel = connection.CreateModel();
        hubContext = _hubContext;
        nodeProcessor = _nodeProcessor;
        healthCheckProcessor = _healthCheckProcessor;
    }

    public void Start(MessageBroker messageBroker)
    {
        var factory = new ConnectionFactory() { HostName = messageBroker.URL }.CreateConnection();

        foreach (Queue queue in messageBroker.Queues)
        {
            channel.QueueDeclare(
                queue: queue.Name,
                durable: queue.Durable,
                exclusive: queue.Exclusive,
                autoDelete: queue.AutoDelete,
                arguments: null
            );

            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);

            consumer.Received += (model, ea) =>
            {
                byte[] body = ea.Body;
                string message = Encoding.UTF8.GetString(body);

                RouteMessage(queue, message);
            };

            channel.BasicConsume(
                    queue: queue.Name,
                    autoAck: queue.AutoAck,
                    consumer: consumer
                );
            hubContext.Clients.All.InvokeAsync("Send", "It worked - from the subscriber");
        }
    }

    public void RouteMessage(Queue queue, string message) {
        if(queue.Name == "discovery") {
            nodeProcessor.Process(message);
        }
        if(queue.Name == "health") {
            healthCheckProcessor.Process(message);
        }
    }

    public void Stop()
    {
        Console.WriteLine("Terminating connection to RabbitMQ instance.");
        channel.Close(200, "Goodbye");
        connection.Close();
    }
}

来自 HealthCheckProcessor.cs

public class HealthCheckProcessor : IHealthCheckProcessor {
    private IDataService dataService;
    private IHubContext<StatusHub> hubContext;

    public HealthCheckProcessor(IDataService _dataService, IHubContext<StatusHub> _hubContext)
    {
        dataService = _dataService;
        hubContext = _hubContext;
    }
    public void Process(string message) {
        HealthCheck health = JsonConvert.DeserializeObject<HealthCheck>(message);
        Node node = dataService.GetSingle(health.NodeId);
        node.Health = health;

        dataService.Update(node);
        Console.WriteLine("It's sending.");
        hubContext.Clients.All.InvokeAsync("Send", "It worked - from the processor");
    }
}

来自 Controller :

[Route("api/[controller]")]
public class MyController: Controller
{
    private IDataService _dataService;
    private readonly IConfiguration configuration;
    private static IHubContext<StatusHub> hubContext;

    public NodesController(IConfiguration config, IDataService dataService, IHubContext<StatusHub> _hubContext)
    {
        _dataService = dataService;
        configuration = config;
        hubContext = _hubContext;
    }

    [HttpGet]
    public string Get()
    {
        hubContext.Clients.All.InvokeAsync("Send", "Blarg!");
        return "Well, I tried.";
    }
}

Answer 1

您正在尝试访问在您请求时不可用的服务。

Configure 在 ConfigureServices 之后调用，以便可以访问任何已注册的服务。

public class Startup {
    public Startup(IConfiguration _configuration) {
        configuration = _configuration;
    }

    public IConfiguration configuration { get; }

    public void ConfigureServices(IServiceCollection services) {        
        services.AddCors();
        services.AddMvc();

        services.AddSignalR();
        services.AddTransient<ISubscriber, Subscriber>();
        services.AddTransient<IDataService, DataService>();
        services.AddTransient<IHealthCheckProcessor, HealthCheckProcessor>();
        services.AddTransient<INodeProcessor, NodeProcessor>();
    }

    public void Configure(
        IApplicationBuilder app, 
        IHostingEnvironment env, 
        IApplicationLifetime applicationLifetime, 
        IServiceProvider sp
    ) {
        if (env.IsDevelopment()) {
            app.UseDeveloperExceptionPage();
        }
        app.UseCors(builder => builder
            // CORS stuff);

        app.UseMvc();

        app.UseSignalR(routes => {
            routes.MapHub<StatusHub>("Status");
        });

        //At this point all the necessary dependencies have been registered and configured
        var subscriber = sp.GetService<ISubscriber>();

        applicationLifetime.ApplicationStarted.Register(() => OnStartup(subscriber));
        applicationLifetime.ApplicationStopping.Register(() => OnShutdown(subscriber));
    }

    private void OnStartup(ISubscriber subscriber) {
        // MessageBroker stuff
        subscriber.Start(messageBroker);
    }

    private void OnShutdown(ISubscriber subscriber) {
        subscriber.Stop();
    }
}

您现在应该能够在构建主机时删除方便的 ConfigureServices。

public static IWebHost BuildWebHost(string[] args) =>
    WebHost.CreateDefaultBuilder(args)
        .UseKestrel()
        .UseIISIntegration()
        .UseStartup<Startup>()
        .Build();