MQTT in ASP.NET Core: Hosted-Service-Consumer mit DI und Health Checks

dotnet new web -n MqttWorker && cd MqttWorker
dotnet add package MQTTnet
dotnet user-secrets init
dotnet user-secrets set "Mqtt:Username" "username"
dotnet user-secrets set "Mqtt:Password" "password"

{
  "Mqtt": {
    "Host": "broker.connectmq.com",
    "Port": 8883,
    "UseTls": true,
    "Topics": [ "testmqtt/aspnet/telemetry/#" ]
  }
}

public sealed class MqttSettings
{
    public string Host { get; set; } = "broker.connectmq.com";
    public int Port { get; set; } = 8883;
    public bool UseTls { get; set; } = true;
    public string? Username { get; set; }
    public string? Password { get; set; }
    public string ClientId { get; set; } = $"aspnet-{Environment.MachineName}";
    public string[] Topics { get; set; } = [];
}

public sealed record InboundMessage(string Topic, string Payload, DateTimeOffset ReceivedAt);

using System.Text.Json;
using System.Text.RegularExpressions;
using System.Threading.Channels;
using Microsoft.AspNetCore.Diagnostics.HealthChecks;
using Microsoft.Extensions.Diagnostics.HealthChecks;
using MQTTnet;
using MQTTnet.Protocol;

var builder = WebApplication.CreateBuilder(args);

builder.Services.Configure<MqttSettings>(builder.Configuration.GetSection("Mqtt"));
builder.Services.AddSingleton<IMqttClient>(_ => new MqttClientFactory().CreateMqttClient());
builder.Services.AddSingleton(_ => Channel.CreateBounded<InboundMessage>(
    new BoundedChannelOptions(1_000) { FullMode = BoundedChannelFullMode.Wait, SingleReader = true }));

builder.Services.AddHostedService<MqttConnectionService>();
builder.Services.AddHostedService<TelemetryConsumer>();
builder.Services.AddScoped<ITelemetryHandler, TelemetryHandler>();

builder.Services.AddHealthChecks()
    .AddCheck<MqttHealthCheck>("mqtt", failureStatus: HealthStatus.Degraded, tags: ["ready"]);

var app = builder.Build();
app.MapHealthChecks("/healthz/live", new HealthCheckOptions { Predicate = _ => false });
app.MapHealthChecks("/healthz/ready", new HealthCheckOptions { Predicate = c => c.Tags.Contains("ready") });
MapCommandEndpoint(app); // lokale Funktion, siehe unten
app.Run();

using System.Threading.Channels;
using Microsoft.Extensions.Options;
using MQTTnet;
using MQTTnet.Formatter;
using MQTTnet.Protocol;

public sealed class MqttConnectionService(
    IMqttClient client,
    IOptions<MqttSettings> settings,
    Channel<InboundMessage> inbox,
    ILogger<MqttConnectionService> log) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var s = settings.Value;
        var builder = new MqttClientOptionsBuilder()
            .WithTcpServer(s.Host, s.Port)
            .WithClientId(s.ClientId)
            .WithCredentials(s.Username, s.Password)
            .WithProtocolVersion(MqttProtocolVersion.V500)
            .WithKeepAlivePeriod(TimeSpan.FromSeconds(30));
        if (s.UseTls) builder.WithTlsOptions(o => o.UseTls());
        var options = builder.Build();

        client.ApplicationMessageReceivedAsync += async e =>
        {
            var m = e.ApplicationMessage;
            // Nur das Nötige kopieren; keine MQTTnet-Objekte über den Handler hinaus halten.
            await inbox.Writer.WriteAsync(
                new InboundMessage(m.Topic, m.ConvertPayloadToString(), DateTimeOffset.UtcNow), stoppingToken);
        };

        client.ConnectedAsync += async _ =>
        {
            var sub = new MqttClientSubscribeOptionsBuilder();
            foreach (var topic in s.Topics) sub.WithTopicFilter(topic, MqttQualityOfServiceLevel.AtLeastOnce);
            var result = await client.SubscribeAsync(sub.Build(), stoppingToken);
            foreach (var item in result.Items)
                log.LogInformation("SUBACK {Topic}: {Code}", item.TopicFilter.Topic, item.ResultCode);
        };

        client.DisconnectedAsync += e =>
        {
            if (e.ClientWasConnected) log.LogWarning(e.Exception, "MQTT disconnected: {Reason}", e.Reason);
            return Task.CompletedTask;
        };

        var delay = TimeSpan.FromSeconds(1);
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                if (!await client.TryPingAsync(stoppingToken))
                {
                    await client.ConnectAsync(options, stoppingToken);
                    log.LogInformation("MQTT connected to {Host}:{Port}", s.Host, s.Port);
                    delay = TimeSpan.FromSeconds(1);
                }
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { break; }
            catch (Exception ex)
            {
                log.LogWarning("MQTT connect failed: {Message}", ex.Message);
                delay = TimeSpan.FromSeconds(Math.Min(delay.TotalSeconds * 2, 60));
            }

            try { await Task.Delay(delay + TimeSpan.FromMilliseconds(Random.Shared.Next(500)), stoppingToken); }
            catch (OperationCanceledException) { break; }
        }
    }

    public override async Task StopAsync(CancellationToken cancellationToken)
    {
        await base.StopAsync(cancellationToken);
        if (client.IsConnected)
            await client.DisconnectAsync(new MqttClientDisconnectOptionsBuilder()
                .WithReason(MqttClientDisconnectOptionsReason.NormalDisconnection).Build(), cancellationToken);
        inbox.Writer.TryComplete();
    }
}

using System.Threading.Channels;

public interface ITelemetryHandler
{
    Task HandleAsync(InboundMessage message, CancellationToken ct);
}

public sealed class TelemetryConsumer(
    Channel<InboundMessage> inbox,
    IServiceScopeFactory scopes,
    ILogger<TelemetryConsumer> log) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await foreach (var msg in inbox.Reader.ReadAllAsync(stoppingToken))
        {
            try
            {
                await using var scope = scopes.CreateAsyncScope();
                var handler = scope.ServiceProvider.GetRequiredService<ITelemetryHandler>();
                await handler.HandleAsync(msg, stoppingToken);
            }
            catch (Exception ex) when (ex is not OperationCanceledException)
            {
                log.LogError(ex, "Failed to handle message on {Topic}", msg.Topic);
            }
        }
    }
}

public sealed class TelemetryHandler(ILogger<TelemetryHandler> log) : ITelemetryHandler
{
    public Task HandleAsync(InboundMessage message, CancellationToken ct)
    {
        log.LogInformation("{Topic}: {Payload}", message.Topic, message.Payload);
        return Task.CompletedTask; // in einer Datenbank speichern, einen anderen Dienst aufrufen, ...
    }
}

using Microsoft.Extensions.Diagnostics.HealthChecks;
using MQTTnet;

public sealed class MqttHealthCheck(IMqttClient client) : IHealthCheck
{
    public Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context, CancellationToken ct = default)
        => Task.FromResult(client.IsConnected
            ? HealthCheckResult.Healthy("Connected to MQTT broker")
            : new HealthCheckResult(context.Registration.FailureStatus, "Not connected to MQTT broker"));
}

static void MapCommandEndpoint(WebApplication app) =>
    app.MapPost("/devices/{id}/commands", async (string id, JsonElement body, IMqttClient mqtt, CancellationToken ct) =>
    {
        if (!Regex.IsMatch(id, "^[A-Za-z0-9_-]{1,64}$")) return Results.BadRequest("invalid device id");
        if (!mqtt.IsConnected) return Results.Problem("MQTT broker unavailable", statusCode: 503);

        var message = new MqttApplicationMessageBuilder()
            .WithTopic($"testmqtt/aspnet/devices/{id}/commands")
            .WithPayload(body.GetRawText())
            .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
            .WithMessageExpiryInterval(60)
            .Build();

        var result = await mqtt.PublishAsync(message, ct);
        return result.ReasonCode is MqttClientPublishReasonCode.Success or MqttClientPublishReasonCode.NoMatchingSubscribers
            ? Results.Accepted()
            : Results.Problem($"Broker rejected publish: {result.ReasonCode}", statusCode: 502);
    });