Broadcast data strategy with .NET Core, SignalR & Angular

· 1 min

Web applications increasingly need to run tasks in the background. Depending on your background, you might call them service workers or background tasks. Here, the web application runs a task for its whole lifetime that continuously emits data for another service to consume.

Because the data never stops, the client side suits an observable pattern: subscribe, and handle new data as it arrives. SignalR pushes each update to every connected client.

First, a little configuration, which is straightforward:

Startup.cs

Add this to ConfigureServices(IServiceCollection services):

services.AddSignalR();

Note: in .NET Core 3 you won't need this call, as the team is tidying up some of the service extensions.

Next, add an endpoint for the hub in Configure(IApplicationBuilder app, IHostingEnvironment env):

app.UseSignalR(routes =>
{
    routes.MapHub<ChangeDataHub>("/changes");
});

ChangeDataHub is a class that inherits from Hub. It can stay empty:

public class ChangeDataHub: Hub
{
}

Next is a listener service that uses the hub manager to send messages to every subscribed client.

First, consider how the listener will run. .NET Core 1 and 2 provided IWebHost, and 2.1 added IHost. IHost also supports hosted services and dependency injection, but it's lighter, because it's designed for background tasks that don't serve HTTP. The data goes out over WebSockets, so it's the right fit.

Registering a hosted service is easy. In Startup.cs, register a singleton that implements IHostedService, an interface with just two methods, both returning Task:

namespace Microsoft.Extensions.Hosting
{
    //
    // Summary:
    //     Defines methods for objects that are managed by the host.
    public interface IHostedService
    {
        //
        // Summary:
        // Triggered when the application host is ready to start the service.
        Task StartAsync(CancellationToken cancellationToken);
        //
        // Summary:
        // Triggered when the application host is performing a graceful shutdown.
        Task StopAsync(CancellationToken cancellationToken);
    }
}

Register the implementation:

public void ConfigureServices(IServiceCollection services)
{
      services.AddSingleton<Microsoft.Extensions.Hosting.IHostedService, ChangeDataListenerService>();
}

Now implement the interface. Like most background tasks, this one shouldn't run once but repeatedly, so it loops with Task.Delay between runs:

private readonly HubLifetimeManager<ChangeDataHub> _hubManager;
private readonly ILogger<DefaultLogger> _logger;
private readonly TimeSpan _delay = TimeSpan.FromSeconds(1);
public ChangeDataListenerService(
    HubLifetimeManager<ChangeDataHub> hubManager,
    ILogger<DefaultLogger> logger)
{
    _hubManager = hubManager;
    _logger = logger;
}
public async Task StartAsync(CancellationToken cts)
{
    _logger.LogInformation("Listener is starting..");
    while (!cts.IsCancellationRequested)
    {
        _logger.LogInformation("Fetching updates..");
        var data = DataManager.Get();
        await _hubManager.SendAllAsync("TransferData", new object[] { data });
        await Task.Delay(_delay, cts);
    }
    _logger.LogInformation("Listener has completed..");
}

That's the basic implementation. It won't work with a repository that uses a DbContext, though: injecting one into a singleton makes the DbContext effectively a singleton too, which leads to a whole host of problems.

The fix is to inject an IServiceScopeFactory and create a scope whenever you need a scoped service:

private readonly HubLifetimeManager<ChangeDataHub> _hubManager;
private readonly ILogger<DefaultLogger> _logger;
private readonly TimeSpan _delay = TimeSpan.FromSeconds(1);
private readonly IServiceScopeFactory _scopeFactory;
public ChangeDataListenerService(
    HubLifetimeManager<ChangeDataHub> hubManager,
    ILogger<DefaultLogger> logger,
    IServiceScopeFactory scopeFactory)
{
    _hubManager = hubManager;
    _logger = logger;
    _scopeFactory = scopeFactory;
}
public async Task StartAsync(CancellationToken cts)
{
    _logger.LogInformation("Listener is starting..");
    while (!cts.IsCancellationRequested)
    {
        // Resolve scoped services and their dependencies in here
        using (var scope = _scopeFactory.CreateScope())
        {
            _logger.LogInformation("Fetching updates..");
            var dataManager = scope.ServiceProvider.GetRequiredService<DataManager>();

            await _hubManager.SendAllAsync("TransferData", new object[] { dataManager.Get() });
        }
        await Task.Delay(_delay, cts);
    }
    _logger.LogInformation("Listener has completed..");
}

It also means the constructor doesn't need every dependency injected, which helps in a larger, real-world service.

The next post covers connecting to this and consuming the data on the client side.