ASP.NET Core MQTT: A Background Service That Stores and Streams Device Data

An ASP.NET Core MQTT integration is the natural backbone for a .NET IoT backend: devices publish to a broker, and your web app subscribes, stores and serves the data. In this guide you will host an MQTTnet subscriber as a BackgroundService, write each reading to a database with EF Core and push it to browsers in real time with SignalR. The finished app runs against broker.connectmq.com for testing and switches to a private ConnectMQ broker through configuration only.

Architecture of an ASP.NET Core MQTT Service

The shape of the solution is simple, and each piece has one job:

Devices --MQTT--> Broker --MQTT--> MqttIngestService (BackgroundService)
                                     |-- EF Core --> SQLite / SQL Server
                                     |-- IHubContext --> SignalR --> Browser dashboard

Why a BackgroundService and not a controller? MQTT is a long-lived, push-based connection. A hosted service starts with the app, keeps one connection open for its lifetime and stops gracefully on shutdown, which is exactly the lifecycle you want. Controllers are request-scoped and would open and close connections constantly.

Two design rules keep this robust:

1. The MQTT client is a singleton owned by the hosted service. Never create a client per request.
2. Database contexts are scoped. The hosted service is a singleton, so it must create a scope per message batch with IServiceScopeFactory.

Project Setup

dotnet new web -n DeviceHub
cd DeviceHub
dotnet add package MQTTnet
dotnet add package Microsoft.EntityFrameworkCore.Sqlite

SignalR is part of the ASP.NET Core shared framework, so no extra server package is needed.

Add the MQTT settings to appsettings.json. For the public broker, use any unique client ID and a random topic prefix; the public area is readable by everyone, so do not send anything sensitive.

{
  "Mqtt": {
    "Host": "broker.connectmq.com",
    "Port": 8883,
    "UseTls": true,
    "ClientId": "devicehub-8d21c4",
    "Username": "",
    "Password": "",
    "TopicFilter": "connectmq/demo/8d21c4/+/telemetry"
  },
  "ConnectionStrings": {
    "Default": "Data Source=devicehub.db"
  }
}

The Data Model and DbContext

using Microsoft.EntityFrameworkCore;

public class Reading
{
    public long Id { get; set; }
    public string DeviceId { get; set; } = "";
    public double Temperature { get; set; }
    public DateTimeOffset ReceivedAt { get; set; }
}

public class DeviceDb(DbContextOptions<DeviceDb> options) : DbContext(options)
{
    public DbSet<Reading> Readings => Set<Reading>();
}

public class MqttOptions
{
    public string Host { get; set; } = "";
    public int Port { get; set; } = 8883;
    public bool UseTls { get; set; } = true;
    public string ClientId { get; set; } = "";
    public string? Username { get; set; }
    public string? Password { get; set; }
    public string TopicFilter { get; set; } = "";
}

The SignalR Hub

The hub itself can be empty. The background service pushes through IHubContext<TelemetryHub>, and browsers only listen.

using Microsoft.AspNetCore.SignalR;

public class TelemetryHub : Hub { }

The MQTT Background Service

This is the heart of the ASP.NET Core MQTT integration. It connects, subscribes, reconnects with backoff and processes messages without blocking MQTTnet's receive loop.

using System.Text.Json;
using System.Threading.Channels;
using Microsoft.AspNetCore.SignalR;
using Microsoft.Extensions.Options;
using MQTTnet;
using MQTTnet.Formatter;
using MQTTnet.Protocol;

public class MqttIngestService(
    IOptions<MqttOptions> options,
    IServiceScopeFactory scopeFactory,
    IHubContext<TelemetryHub> hub,
    ILogger<MqttIngestService> logger) : BackgroundService
{
    private readonly MqttOptions _opt = options.Value;
    private readonly Channel<Reading> _queue =
        Channel.CreateBounded<Reading>(new BoundedChannelOptions(10_000)
        { FullMode = BoundedChannelFullMode.DropOldest });

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var client = new MqttClientFactory().CreateMqttClient();
        client.ApplicationMessageReceivedAsync += OnMessageAsync;

        var builder = new MqttClientOptionsBuilder()
            .WithTcpServer(_opt.Host, _opt.Port)
            .WithClientId(_opt.ClientId)
            .WithProtocolVersion(MqttProtocolVersion.V500)
            .WithCleanStart(false)
            .WithSessionExpiryInterval(1800);
        if (_opt.UseTls) builder.WithTlsOptions(o => o.UseTls());
        if (!string.IsNullOrEmpty(_opt.Username))
            builder.WithCredentials(_opt.Username, _opt.Password);
        var mqttOptions = builder.Build();

        var writer = Task.Run(() => WriteLoopAsync(stoppingToken), stoppingToken);
        var delay = TimeSpan.FromSeconds(1);

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                if (!await client.TryPingAsync(stoppingToken))
                {
                    var result = await client.ConnectAsync(mqttOptions, stoppingToken);
                    logger.LogInformation("MQTT connected, session present: {Present}",
                        result.IsSessionPresent);
                    if (!result.IsSessionPresent)
                    {
                        await client.SubscribeAsync(new MqttClientSubscribeOptionsBuilder()
                            .WithTopicFilter(f => f.WithTopic(_opt.TopicFilter)
                                .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce))
                            .Build(), stoppingToken);
                    }
                    delay = TimeSpan.FromSeconds(1);
                }
            }
            catch (OperationCanceledException) { break; }
            catch (Exception ex)
            {
                logger.LogWarning("MQTT connect failed: {Message}", ex.Message);
                delay = TimeSpan.FromSeconds(Math.Min(delay.TotalSeconds * 2, 60));
            }

            try { await Task.Delay(delay, stoppingToken); }
            catch (OperationCanceledException) { break; }
        }

        if (client.IsConnected) await client.DisconnectAsync();
        client.Dispose();
        _queue.Writer.TryComplete();
        await writer;
    }

    private Task OnMessageAsync(MqttApplicationMessageReceivedEventArgs e)
    {
        try
        {
            // Topic: connectmq/demo/<prefix>/<deviceId>/telemetry
            var parts = e.ApplicationMessage.Topic.Split('/');
            var deviceId = parts[^2];
            using var doc = JsonDocument.Parse(e.ApplicationMessage.ConvertPayloadToString());
            var reading = new Reading
            {
                DeviceId = deviceId,
                Temperature = doc.RootElement.GetProperty("temperature").GetDouble(),
                ReceivedAt = DateTimeOffset.UtcNow
            };
            _queue.Writer.TryWrite(reading);
        }
        catch (Exception ex)
        {
            logger.LogWarning("Dropping bad payload on {Topic}: {Message}",
                e.ApplicationMessage.Topic, ex.Message);
        }
        return Task.CompletedTask;
    }

    private async Task WriteLoopAsync(CancellationToken ct)
    {
        var batch = new List<Reading>(100);
        await foreach (var first in _queue.Reader.ReadAllAsync(CancellationToken.None))
        {
            batch.Add(first);
            while (batch.Count < 100 && _queue.Reader.TryRead(out var next))
                batch.Add(next);

            using (var scope = scopeFactory.CreateScope())
            {
                var db = scope.ServiceProvider.GetRequiredService<DeviceDb>();
                db.Readings.AddRange(batch);
                await db.SaveChangesAsync(CancellationToken.None);
            }

            foreach (var r in batch)
                await hub.Clients.All.SendAsync("reading", r, CancellationToken.None);

            batch.Clear();
        }
    }
}

A few choices worth explaining:

• The message handler only parses and enqueues. Database and SignalR calls happen in a separate loop. If the handler awaited a slow database, MQTTnet would stop reading the socket, keep-alives could time out and the broker would drop the connection.
• A bounded channel with DropOldest caps memory if the database goes down. For telemetry, losing the oldest readings is usually better than crashing the process. If you cannot lose data, switch to Wait and accept backpressure.
• Batching turns hundreds of inserts into one SaveChangesAsync call.
• Persistent session (CleanStart(false) plus a 30-minute session expiry) means QoS 1 messages published while your app restarts are delivered when it reconnects, and the subscription survives. That only works if the client ID stays stable, which is why it lives in configuration.
• Malformed payloads are logged and dropped, never rethrown. One bad device should not take down ingestion.

Wiring It Up in Program.cs

using Microsoft.EntityFrameworkCore;

var builder = WebApplication.CreateBuilder(args);

builder.Services.Configure<MqttOptions>(builder.Configuration.GetSection("Mqtt"));
builder.Services.AddDbContext<DeviceDb>(o =>
    o.UseSqlite(builder.Configuration.GetConnectionString("Default")));
builder.Services.AddSignalR();
builder.Services.AddHostedService<MqttIngestService>();

var app = builder.Build();

using (var scope = app.Services.CreateScope())
    scope.ServiceProvider.GetRequiredService<DeviceDb>().Database.EnsureCreated();

app.UseDefaultFiles();
app.UseStaticFiles();
app.MapHub<TelemetryHub>("/hubs/telemetry");

app.MapGet("/api/readings/{deviceId}", async (string deviceId, DeviceDb db) =>
    await db.Readings.Where(r => r.DeviceId == deviceId)
        .OrderByDescending(r => r.Id).Take(50).ToListAsync());

app.Run();

EnsureCreated is fine for a demo; use EF Core migrations in a real project.

A Minimal Live Dashboard

Create wwwroot/index.html with a list and the SignalR JavaScript client from a CDN. The essential script is:

const connection = new signalR.HubConnectionBuilder()
  .withUrl("/hubs/telemetry")
  .withAutomaticReconnect()
  .build();

connection.on("reading", r => {
  const li = document.createElement("li");
  li.textContent = `${r.receivedAt} ${r.deviceId}: ${r.temperature} °C`;
  document.getElementById("log").prepend(li);
});

connection.start();

Note the camelCase property names: SignalR's JSON protocol serializes C# properties in camelCase by default.

Test the Pipeline End to End

Run the app with dotnet run, open the page, then publish a fake reading from a terminal:

mosquitto_pub -h broker.connectmq.com -p 1883 -i devicehub-test-1 \
  -t connectmq/demo/8d21c4/boiler-1/telemetry -q 1 \
  -m '{"temperature": 71.4}'

The reading should appear in the browser within a fraction of a second and be retrievable at /api/readings/boiler-1. For more command-line variations, see the mosquitto_pub and mosquitto_sub examples.

Sending Commands Back to Devices

Ingestion is only half of most IoT backends; the other half is control. A common mistake is to open a new MQTT connection inside an HTTP endpoint for every command. That is slow, wastes broker connections (which count against your plan's quota) and fails under load. Instead, keep a second long-lived client for publishing, or expose the ingestion client through a small singleton wrapper:

public class MqttCommandPublisher(IMqttClient client)
{
    public Task<MqttClientPublishResult> SendAsync(string deviceId, object command,
        CancellationToken ct) =>
        client.PublishAsync(new MqttApplicationMessageBuilder()
            .WithTopic($"connectmq/demo/8d21c4/{deviceId}/cmd")
            .WithPayload(JsonSerializer.Serialize(command))
            .WithContentType("application/json")
            .WithMessageExpiryInterval(30)
            .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
            .Build(), ct);
}

The message expiry interval matters for commands: if a device is offline for ten minutes, you almost never want it to execute a "open valve" command the moment it reconnects. With a 30-second expiry, the broker discards the stale command instead of delivering it. Check the reason code in the returned MqttClientPublishResult and surface failures to the caller, rather than returning 200 OK for a command that was never accepted.

Whether you share one client between ingestion and commands or use two is a judgment call. Two clients give clearer logs and independent reconnect behavior; one client saves a connection. On a small plan, one is usually fine.

Observability

A background service fails quietly unless you make it loud. At minimum, log connection state changes with the reason, count messages received and dropped per minute, and record the queue depth of the channel. ASP.NET Core's built-in ILogger and a health check that reports IsConnected cover most needs; if you already use OpenTelemetry, expose those counters as metrics so you can alert when ingestion stops while devices are still publishing.

Production Checklist

Shared subscriptions and QoS 2 are available on the paid Serverless plan and above, so plan for that before running more than one ingestion instance. The MQTT 5 features guide explains shared subscriptions and session expiry in more detail.

Switching to a Private Broker

Production devices should talk to a private, authenticated broker. ConnectMQ provides a fully managed MQTT service with TLS on 8883, MQTT 3.1.1 and 5.0, and per-plan quotas, so your ASP.NET Core app only has to change configuration:

dotnet user-secrets set "Mqtt:Host" "YOUR_BROKER_HOST"
dotnet user-secrets set "Mqtt:ClientId" "YOUR_CLIENT_ID"
dotnet user-secrets set "Mqtt:Username" "YOUR_USERNAME"
dotnet user-secrets set "Mqtt:Password" "YOUR_PASSWORD"

Because ConnectMQ binds client IDs to issued credentials, use the client ID generated in the portal for the backend, and give each device its own credentials. If you want to understand the client options in more depth, the MQTTnet tutorial covers TLS, MQTT 5 properties and reconnect details, and the MQTT client libraries guide compares .NET with other ecosystems. If you are evaluating where to host the broker at all, read about running an MQTT broker in the cloud.

Get Started with ConnectMQ

Create a free Serverless broker at app.connectmq.com, no credit card required, drop the generated credentials into user secrets and your ASP.NET Core MQTT service is production-shaped from day one. Use the portal's live topic explorer to watch device traffic while you develop the backend.

Frequently Asked Questions

Should I use IHostedService or BackgroundService for MQTT?

BackgroundService is the simpler choice because it gives you a single ExecuteAsync method with a stopping token. It implements IHostedService for you, so both register the same way with AddHostedService.

Why not inject DbContext directly into the background service?

Hosted services are singletons and DbContext is scoped. Injecting it directly either fails validation or keeps one context alive forever, so create a scope per batch with IServiceScopeFactory.

How do I run several instances without processing each message twice?

Use an MQTT 5 shared subscription such as $share/devicehub/connectmq/+/telemetry. The broker then delivers each message to only one member of the group.

Can I publish commands to devices from a controller?

Yes. Register the connected IMqttClient (or a small wrapper) as a singleton and call PublishAsync from your endpoint; do not create a new connection per HTTP request.