# MQTT in .NET

You may have recently seen the news about hackers changing refrigeration settings in French supermarkets, causing temperatures to rise and forcing stores to throw away spoiled food. Its a good example of how software controlling physical equipment can have very tangible consequences. Now imagine building the temperature monitoring system for one of those supermarkets. Each sensor sends a reading every few seconds. A dashboard displays it, an alerting service checks whether the fridge is getting too warm, and a storage service records it for later analysis. Three applications need the same measurement, but the sensor shouldnt need three HTTP clients, three sets of endpoint configuration, and three separate retry policies to keep them informed.

MQTT gives that sensor one connection to a broker. It publishes a message to a topic, and the broker forwards the message to interested subscribers. Adding another consumer does not require changing the sensor. For .NET developers, the first connection is straightforward. The more interesting work starts when a subscriber disconnects, a reading arrives twice, or a successful publish is mistaken for proof that something happened in the database. This post builds a small example with MQTTnet, then follows those failure cases into the application design.

## What MQTT actually does

MQTT is a lightweight publish/subscribe messaging protocol. Clients normally maintain a TCP connection to a broker, which routes messages according to topic subscriptions. A client can publish and subscribe over the same connection. The payload is application defined, JSON is convenient, but MQTT does not require it. The broker is the running server. MQTT is the protocol it speaks. MQTTnet is the .NET library we will use to speak that protocol. Keeping those three separate makes the architecture easier to picture.

Imagine that our application monitors refrigeration equipment in a warehouse. Each fridge publishes temperature readings. A .NET ingestion service stores them, while a separate service watches for readings outside the permitted range.

![](https://cdn.hashnode.com/uploads/covers/67c36038c69a4b7143c5fc49/49c1a0e3-c237-4063-99d7-3714458307dd.png align="center")

The sensor knows where the broker is and which topic to publish to. It doesnt know which services are running behind the broker. That is useful when devices are deployed in places where updating them is slower and more expensive than deploying another backend service. The [MQTT 5 specification](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html) defines the protocol. Broker deployment, database transactions, and application retry policies remain separate engineering decisions.

## Topics become part of your contract

A topic is a case sensitive routing name. For our fridge, we could use:

```text
warehouse/dublin/devices/fridge-17/telemetry/temperature
```

A subscriber can use that exact topic or a topic filter containing wildcards. The `+` wildcard matches one level, so this filter receives temperature messages for all devices at the Dublin warehouse:

```text
warehouse/dublin/devices/+/telemetry/temperature
```

The `#` wildcard matches multiple levels and must appear at the end of the filter. This receives everything beneath the Dublin warehouse:

```text
warehouse/dublin/#
```

Publishers send to concrete topics. Wildcards belong in subscription filters. HiveMQ's [topic guide](https://www.hivemq.com/blog/mqtt-essentials-part-5-mqtt-topics-best-practices/) explains these matching rules in more detail. Treat the topic structure as an API contract. Renaming a level can break consumers just as changing an HTTP route can. I would keep tenant or site boundaries explicit, use stable device identifiers, and separate telemetry from commands and status. For example, `telemetry/temperature`, `commands/set-target`, and `status` serve different purposes. They should not all share one topic with a payload field that every subscriber has to inspect before deciding whether a message is relevant. Avoid putting a timestamp or message identifier into each topic. Those values belong in the payload. Otherwise, a straightforward stream of readings turns into an ever growing collection of topic names, and wildcard permissions become harder to reason about.

## Start a local broker

We will use Eclipse Mosquitto locally. Create a directory containing `compose.yaml` and `mosquitto.conf`.

```yaml
# compose.yaml
services:
  mqtt:
    image: eclipse-mosquitto:2
    ports:
      - "127.0.0.1:1883:1883"
    volumes:
      - ./mosquitto.conf:/mosquitto/config/mosquitto.conf:ro
```

```text
# mosquitto.conf
listener 1883
allow_anonymous true
persistence false
```

Start it from that directory:

```bash
docker compose up -d
```

The host port is bound to loopback, and this broker deliberately permits anonymous connections. It is a disposable development setup. It also has persistence disabled, so restarting it is not a durability test. The [Mosquitto configuration reference](https://mosquitto.org/man/mosquitto-conf-5.html) documents listener, authentication, persistence, and queue settings. The .NET application below runs on the host and connects to `localhost`. If you later put it in the same Compose network, use `mqtt` as the hostname instead. Inside an application container, `localhost` refers to that container.

## Build a publisher and subscriber

This example targets .NET 10 and MQTTnet 5. It uses a single console project with two modes, so both ends share the same message contract. The package version is pinned to make the example reproducible.

```bash
dotnet new console --framework net10.0 --name FridgeMqtt
cd FridgeMqtt
dotnet add package MQTTnet --version 5.2.0.1603
```

MQTTnet 5 uses `MqttClientFactory`. Older tutorials often use `MqttFactory` and namespaces from previous major versions, so check the [upgrade guide](https://github.com/dotnet/MQTTnet/wiki/Upgrading-guide) before combining examples. The [NuGet version listing](https://www.nuget.org/packages/MQTTnet) and official [connection](https://github.com/dotnet/MQTTnet/blob/master/Samples/Client/Client_Connection_Samples.cs), [publishing](https://github.com/dotnet/MQTTnet/blob/master/Samples/Client/Client_Publish_Samples.cs), and [subscription](https://github.com/dotnet/MQTTnet/blob/master/Samples/Client/Client_Subscribe_Samples.cs) samples are useful references.

Replace `Program.cs` with the following code:

```csharp
using System.Text.Json;
using MQTTnet;
using MQTTnet.Formatter;
using MQTTnet.Protocol;

if (args.Length != 1 || args[0] is not ("publish" or "subscribe"))
{
    Console.WriteLine("Usage: dotnet run -- publish|subscribe");
    return;
}

var mode = args[0];
const string topic =
    "warehouse/dublin/devices/fridge-17/telemetry/temperature";
const string topicFilter =
    "warehouse/dublin/devices/+/telemetry/temperature";

using var stopSource = new CancellationTokenSource();
var stopToken = stopSource.Token;

Console.CancelKeyPress += (_, eventArgs) =>
{
    eventArgs.Cancel = true;
    stopSource.Cancel();
};

var factory = new MqttClientFactory();
using var client = factory.CreateMqttClient();
client.DisconnectedAsync += eventArgs =>
{
    if (!stopToken.IsCancellationRequested)
    {
        Console.Error.WriteLine($"Disconnected: {eventArgs.Reason}");
        stopSource.Cancel();
    }
    return Task.CompletedTask;
};

if (mode == "subscribe")
{
    client.ApplicationMessageReceivedAsync += eventArgs =>
    {
        try
        {
            var reading = JsonSerializer.Deserialize<TemperatureReading>(
                eventArgs.ApplicationMessage.ConvertPayloadToString());

            if (reading is null || reading.SchemaVersion != 1 ||
                reading.MessageId == Guid.Empty ||
                string.IsNullOrWhiteSpace(reading.DeviceId) ||
                !double.IsFinite(reading.Celsius))
            {
                Console.Error.WriteLine("Rejected invalid reading.");
                return Task.CompletedTask;
            }

            Console.WriteLine(
                $"{reading.RecordedAtUtc:O} " +
                $"{reading.DeviceId}: {reading.Celsius:F1} °C " +
                $"[{reading.MessageId}]");
        }
        catch (JsonException exception)
        {
            Console.Error.WriteLine($"Invalid JSON: {exception.Message}");
        }
        return Task.CompletedTask;
    };
}

var options = new MqttClientOptionsBuilder()
    .WithClientId($"fridge-demo-{mode}-{Guid.NewGuid():N}")
    .WithTcpServer("localhost", 1883)
    .WithProtocolVersion(MqttProtocolVersion.V500)
    .WithCleanSession()
    .Build();

try
{
    var connection = await client.ConnectAsync(options, stopToken);

    if (connection.ResultCode != MqttClientConnectResultCode.Success)
        throw new InvalidOperationException(
            $"Connection rejected: {connection.ResultCode}");

    if (mode == "subscribe")
    {
        var subscription = factory.CreateSubscribeOptionsBuilder()
            .WithTopicFilter(filter => filter
                .WithTopic(topicFilter)
                .WithQualityOfServiceLevel(
                    MqttQualityOfServiceLevel.AtLeastOnce))
            .Build();

        var result = await client.SubscribeAsync(subscription, stopToken);
        if (result.Items.Any(item =>
            item.ResultCode != MqttClientSubscribeResultCode.GrantedQoS1))
        {
            throw new InvalidOperationException(
                "The broker did not grant the requested QoS 1 subscription.");
        }
        Console.WriteLine("Subscribed. Press Ctrl+C to stop.");
        await Task.Delay(Timeout.InfiniteTimeSpan, stopToken);
    }
    else
    {
        var deviceRunId = Guid.NewGuid();
        long sequence = 0;
        while (!stopToken.IsCancellationRequested)
        {
            var reading = new TemperatureReading(
                SchemaVersion: 1,
                MessageId: Guid.NewGuid(),
                DeviceId: "fridge-17",
                DeviceRunId: deviceRunId,
                Sequence: ++sequence,
                RecordedAtUtc: DateTimeOffset.UtcNow,
                Celsius: Math.Round(2 + Random.Shared.NextDouble() * 3, 1));

            var message = new MqttApplicationMessageBuilder()
                .WithTopic(topic)
                .WithPayload(JsonSerializer.SerializeToUtf8Bytes(reading))
                .WithContentType("application/json")
                .WithQualityOfServiceLevel(
                    MqttQualityOfServiceLevel.AtLeastOnce)
                .WithRetainFlag(false)
                .Build();

            var result = await client.PublishAsync(message, stopToken);
            if ((int)result.ReasonCode >= 0x80)
                throw new InvalidOperationException(
                    $"Publish rejected: {result.ReasonCode}");

            Console.WriteLine(
                $"Published {reading.MessageId}: {result.ReasonCode}");
            await Task.Delay(TimeSpan.FromSeconds(2), stopToken);
        }
    }
}
catch (OperationCanceledException) when (stopToken.IsCancellationRequested)
{
    // Ctrl+C or a lost connection ends 
}
finally
{
    if (client.IsConnected)
    {
        using var disconnectSource =
            new CancellationTokenSource(TimeSpan.FromSeconds(5));
        try
        {
            await client.DisconnectAsync(
                factory.CreateClientDisconnectOptionsBuilder().Build(),
                disconnectSource.Token);
        }
        catch (Exception exception)
        {
            Console.Error.WriteLine($"Disconnect failed: {exception.Message}");
        }
    }
}

public sealed record TemperatureReading(
    int SchemaVersion,
    Guid MessageId,
    string DeviceId,
    Guid DeviceRunId,
    long Sequence,
    DateTimeOffset RecordedAtUtc,
    double Celsius);
```

Start the subscriber in one terminal:

```bash
dotnet run -- subscribe
```

Once it prints `Subscribed`, run the publisher in another terminal from the same project directory:

```bash
dotnet run --no-build -- publish
```

The subscriber should print a reading roughly every two seconds. Stop it and start it again while the publisher keeps running. It will receive new readings after subscribing, but it will not recover the intervening readings. That is deliberate: the demonstration uses fresh client identities, clean sessions, and non retained messages. These examples have been checked against the documented MQTTnet 5 API patterns, but have not been compiled or run as part of this article's preparation.

## Give messages an identity

The temperature value is the smallest part of the contract. `MessageId` identifies the logical reading, `SchemaVersion` describes the payload shape, and `RecordedAtUtc` says when the measurement was taken. `DeviceRunId` and `Sequence` let a consumer detect gaps within one publisher run without confusing a restarted sequence with an earlier reading. For real devices, decide whether sequence numbers survive a reboot. If they do not, pair them with a boot identifier as the example does. Do not assume that every device clock is accurate enough to establish a global order. Most importantly, preserve `MessageId` when retrying the same logical reading. Generating a new identifier for every publish attempt makes two attempts look like two separate measurements. An inbox cannot deduplicate them if the producer has already discarded their shared identity. The subscribers validation is intentionally small. A production ingestion boundary should also validate the device against the topic, enforce payload limits, check supported units, and apply sensible timestamp and range rules. Authentication of the connection does not make every field in its JSON trustworthy.

## What QoS guarantees

MQTT has three quality of service levels. QoS 0 sends without an MQTT acknowledgement and allows loss. QoS 1 adds acknowledgement and permits duplicate delivery. QoS 2 uses an additional protocol exchange to provide exactly-once delivery within the MQTT exchange. The publisher-to-broker and broker-to-subscriber deliveries are separate exchanges, a subscribers granted QoS can lower its delivery level. See the [QoS guide](https://www.hivemq.com/blog/mqtt-essentials-part-6-mqtt-quality-of-service-levels/). For this example, QoS 1 gives us a useful place to start, provided the application can tolerate duplicates. Choosing QoS 2 would not create a transaction spanning the broker, the subscriber, and SQL Server.

With a subscriber that inserts a reading into a database. It commits successfully, then crashes before its acknowledgement reaches the broker. After recovery, it may receive that reading again. If the handler blindly inserts another row or sends another alert, the business effect can be duplicated.

![](https://cdn.hashnode.com/uploads/covers/67c36038c69a4b7143c5fc49/9b3a2481-c753-4c53-8086-927d0e23a877.png align="center")

That diagram assumes a recoverable session and retained delivery state, our disposable console demo does not configure them. It illustrates the failure case the production design needs to address. There is a similar distinction on the publishing side. A successful publish acknowledgement tells the publisher about the broker exchange. It does not prove that an alert was sent, that any particular subscriber was connected, or that a downstream database committed. If the sender needs confirmation of a business outcome, model that outcome explicitly with a correlated application response.

## Put the acknowledgement after durable work

For ingestion that must survive process crashes, I would introduce a durable inbox. The worker first records the message identity and payload in a transaction, protected by a unique constraint. It acknowledges the MQTT message after that transaction commits. A separate processing step can then update projections or produce alerts from the inbox. Another valid design performs the business update and records the processed identity in the same transaction, then acknowledges. The important property is that the duplicate check and the protected database changes commit together. This is application design rather than something MQTT supplies automatically. A read before insert check is insufficient under concurrent delivery: two handlers can both observe that the identity is absent. The database needs to arbitrate the race with a unique constraint, and the code needs to recognise that specific duplicate key outcome.

MQTTnet exposes `AutoAcknowledge` and `AcknowledgeAsync` for explicit acknowledgement control. The official [subscription samples](https://github.com/dotnet/MQTTnet/blob/master/Samples/Client/Client_Subscribe_Samples.cs) demonstrate these APIs. Disable automatic acknowledgement before transferring responsibility to asynchronous work, and acknowledge only when the durable handoff has succeeded. Returning from a callback after placing a message into an in-memory `Channel<T>` leaves a crash window if that return permits acknowledgement. The channel can organise work and bound concurrency, but its contents disappear with the process. Also avoid treating an unacknowledged message as a generic delayed retry. MQTT does not provide a queue style abandon operation that automatically schedules another application attempt. Your design needs a defined recovery path, including session resumption where applicable and a policy for permanently invalid messages. A durable quarantine record followed by acknowledgement can be more useful than repeatedly blocking on the same malformed payload.

## Retained messages describe current state

A retained publication lets the broker keep a latest value for a topic so that a later subscriber can receive it. Publishing another retained value replaces the previous one, a retained zero byte payload clears it. It is not a history of all earlier publications. See the [retained-message guide](https://www.hivemq.com/blog/mqtt-essentials-part-8-retained-messages/). For the fridge, I would keep the event stream used for historical ingestion separate from a retained current state topic:

```text
warehouse/dublin/devices/fridge-17/telemetry/temperature
warehouse/dublin/devices/fridge-17/state/temperature
```

The historical consumer handles readings as events. A dashboard can subscribe to the state topic and show the last reported value immediately. Its UI should still display the measurement time, because a retained value can be old. Retaining imperative commands requires much more care. A late subscriber receiving an old `defrost-now` message should not trigger a new defrost cycle. Desired state such as `target-temperature = 4` is easier to reconcile repeatedly, especially when the contract includes a version and the device reports the version it has applied.

## Reconnection needs a session policy

Reopening a socket and recovering missed work are different requirements. A reconnect loop can restore connectivity while still losing everything published during the outage. MQTT 5 separates Clean Start from Session Expiry Interval. A client that wants to resume a session uses a stable client identity, requests resumption, and configures a non-zero expiry interval. It must check the broker's Session Present response and restore subscriptions when no session was resumed. Session retention and offline delivery are also constrained by broker limits and message expiry. The [MQTT 5 specification](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html) defines those controls, the [persistent-session overview](https://www.hivemq.com/blog/mqtt-essentials-part-7-persistent-session-queuing-messages/) explains the underlying use case.

The stable identity belongs to one logical client. Giving two running replicas the same client ID can cause them to displace one another. Random identifiers solve collisions but prevent those replicas from resuming their old sessions. Deployment identity and session ownership therefore need an explicit design. For a .NET worker, I would keep connection management in one supervised loop with bounded exponential backoff and jitter. One owner should decide when to connect, subscribe, and disconnect. Reconnecting independently from several event handlers makes it much harder to know which subscriptions are active.

I would also distinguish an unavailable broker from rejected credentials. Retrying a temporary outage is reasonable. Repeating a bad password indefinitely should produce a visible configuration failure. Finally, broker side session state cannot preserve a measurement that never left the device. If every reading must survive a network outage and a device restart, the publisher needs a durable local spool, a capacity limit, and a defined policy for what happens when that capacity is exhausted.

## Report presence with a Last Will

MQTT lets a client register a Last Will when connecting. The broker can publish that message when the connection is lost abnormally. A device can register an `offline` will on its status topic and publish `online` after connecting, with retained status helping later subscribers. The [Last Will guide](https://www.hivemq.com/blog/mqtt-essentials-part-9-last-will-and-testament/) covers the mechanism. Presence is still an observation. A broken network path can make a healthy device appear offline, and a connected device can have a broken sensor. I would display connectivity separately from the age of the last valid reading. A normal shutdown also needs an explicit status decision. If the client disconnects gracefully without publishing an offline state, a retained online value can remain misleading. Publish the intended shutdown status before the normal disconnect, while allowing a bounded shutdown timeout.

## Scale consumers deliberately

If three workers each create an ordinary subscription to the same topic filter, each can receive a copy. That is useful when each worker serves a different purpose. It is usually surprising when all three are replicas of the same ingestion service. Shared subscriptions allow instances to join a named group so that a matching publication is assigned to one group member. With a supporting broker, the filter could be:

```text
$share/temperature-ingestion/warehouse/+/devices/+/telemetry/temperature
```

A separate alerting group can receive its own stream while ingestion replicas share theirs. The [shared-subscription guide](https://www.hivemq.com/blog/mqtt5-essentials-part7-shared-subscriptions/) describes this distribution model. Do not infer per device processing order from membership of a shared group. Two consecutive readings may be processed by different workers at different speeds. A projection that stores the latest state should reject an obsolete version rather than allowing a slow handler to overwrite a newer reading. The device run identifier and sequence in our contract help identify that situation, though switching between device runs still requires a defined policy. Scaling also needs backpressure. If arrivals exceed database capacity for long enough, adding an unbounded collection merely moves the outage into memory. Bound in process work, monitor backlog age, and decide which data can be dropped, compacted, or durably queued. Current temperature displays and compliance archives can reasonably make different choices.

## Host the connection with the application

In a .NET service, an MQTT connection usually belongs to application lifetime. A `BackgroundService` can own the connection and subscription lifecycle. Avoid connecting and disconnecting for every HTTP request just because the request initiated a publication. If the message handler uses EF Core, create a dependency injection scope for each bounded unit of processing. Do not hold a scoped `DbContext` inside a singleton worker or share one context between concurrent handlers. Connection lifetime and database unit of work lifetime are different.

For an HTTP endpoint that must reliably initiate a device command, I would normally store the command and an outbox entry in the same database transaction. A dispatcher then publishes the command with its original command ID. If the broker receives it but the dispatcher crashes before marking the outbox entry complete, publication may repeat; the device must recognise that command ID.

![](https://cdn.hashnode.com/uploads/covers/67c36038c69a4b7143c5fc49/edc1e5db-d1bf-493d-9395-e8965cf97d44.png align="center")

The API can report that a command has been accepted and expose its status separately. Accepted, published, received, and applied are useful distinct states. An HTTP success response should reflect the state actually achieved.

## Secure the topic boundary

The local broker was intentionally anonymous. A deployed system needs authenticated clients, encrypted connections, and authorisation for the topics each identity can publish to or subscribe to. A fridge should be able to publish its own telemetry and receive its own commands. It should not be able to publish another device's status or subscribe to every tenant's command stream. A well structured topic hierarchy makes those permissions easier to express, but the hierarchy itself enforces nothing. Keep certificate validation enabled when configuring TLS in MQTTnet. If a private certificate authority is involved, configure trust correctly rather than accepting every certificate. Credentials should come from deployment configuration or a secret store, with rotation handled as an operational requirement. The ingestion service should also compare the payloads claimed device identity with the authorised topic it arrived on. Otherwise, an authorised publisher can send a plausible looking JSON document that attributes its data to somebody else.

## Test the failures you depend on

The happy path console output proves very little about durability. To evaluate the production design, stop a subscriber while readings continue and verify exactly which readings return after reconnection. Restart the broker and check whether its configured persistence preserves the state you depend on. Kill the worker after its database commit but before acknowledgement and verify that redelivery produces one business effect. Also publish malformed JSON and an unsupported schema version. A useful system records why those messages were rejected without letting one poison message stall unrelated devices. Slow the database deliberately and observe whether memory stays bounded and backlog age becomes visible.

I would monitor connection failures, rejected subscriptions, publish failures, durable ingestion latency, duplicate detections, quarantine counts, and the age of the oldest unprocessed record. A worker can have a healthy MQTT connection while its database work has stopped making progress. For the basic example, restarting the broker should stop both console processes because they cancel on disconnection. That behaviour is intentionally visible. Replacing it with reconnection is the next implementation step, alongside deciding what recovery must preserve.

## Where MQTT belongs

MQTT fits naturally at a device boundary where many independently deployed clients send small updates and receive commands over long lived connections. In that architecture, .NET can provide ingestion, validation, persistence, command tracking, and the APIs used by people and other services. For an ordinary backend workflow, choose the transport around the operational features the workflow requires. If delayed retries, durable work queues, dead letter handling, or replay are central requirements, evaluate how the chosen broker and application framework supply them. MQTT support alone does not answer those questions.

The practical starting point is small, a broker, a topic contract, a publisher, and a subscriber. From there, make the delivery boundaries explicit. Decide when a reading becomes durable, how a duplicate is recognised, how a disconnected client recovers, and how a command's outcome reaches its caller. Those decisions turn a successful message exchange into a solid system.