CloudAMQP with MQTT and .NET

We recommend MQTTnet, the .NET Foundation’s MQTT client. The examples are .NET 10 file-based apps: dotnet run installs the MQTTnet package their #:package line names. In a project, add it with:

dotnet add package MQTTnet

Set MQTT_HOST, MQTT_PORT, MQTT_USERNAME and MQTT_PASSWORD to the MQTT details on your instance’s details page in the CloudAMQP console. Port 8883 uses TLS. Without them, the examples connect to a local broker.

Save the publisher as publisher.cs:

#:package MQTTnet@5.2.0.1603
using MQTTnet;
using MQTTnet.Formatter;
using MQTTnet.Protocol;

var port = int.Parse(Environment.GetEnvironmentVariable("MQTT_PORT") ?? "1883");
var options = new MqttClientOptionsBuilder()
    .WithTcpServer(Environment.GetEnvironmentVariable("MQTT_HOST") ?? "localhost", port)
    .WithCredentials(Environment.GetEnvironmentVariable("MQTT_USERNAME") ?? "guest", Environment.GetEnvironmentVariable("MQTT_PASSWORD") ?? "guest")
    .WithTlsOptions(tls => tls.UseTls(port == 8883))
    // MQTTnet defaults to MQTT 5, LavinMQ supports MQTT 3.1.1
    .WithProtocolVersion(MqttProtocolVersion.V311)
    .Build();
using var client = new MqttClientFactory().CreateMqttClient();
await client.ConnectAsync(options);

// QoS 1 completes once the broker has sent its PUBACK
await client.PublishStringAsync("hello", "Hello world!", MqttQualityOfServiceLevel.AtLeastOnce);
Console.WriteLine(" [x] Sent 'Hello world!'");
await client.DisconnectAsync();

Save the subscriber as worker.cs:

#:package MQTTnet@5.2.0.1603
using MQTTnet;
using MQTTnet.Formatter;
using MQTTnet.Protocol;

var port = int.Parse(Environment.GetEnvironmentVariable("MQTT_PORT") ?? "1883");
var options = new MqttClientOptionsBuilder()
    .WithTcpServer(Environment.GetEnvironmentVariable("MQTT_HOST") ?? "localhost", port)
    .WithCredentials(Environment.GetEnvironmentVariable("MQTT_USERNAME") ?? "guest", Environment.GetEnvironmentVariable("MQTT_PASSWORD") ?? "guest")
    .WithTlsOptions(tls => tls.UseTls(port == 8883))
    // MQTTnet defaults to MQTT 5, LavinMQ supports MQTT 3.1.1
    .WithProtocolVersion(MqttProtocolVersion.V311)
    .Build();
using var client = new MqttClientFactory().CreateMqttClient();
await client.ConnectAsync(options);

// Each message is acked when the handler's task completes
client.ApplicationMessageReceivedAsync += e =>
{
    Console.WriteLine($" [x] Received '{e.ApplicationMessage.ConvertPayloadToString()}'");
    return Task.CompletedTask;
};
await client.SubscribeAsync("hello", MqttQualityOfServiceLevel.AtLeastOnce);
Console.WriteLine(" [*] Waiting for messages. To exit press CTRL+C");
await Task.Delay(Timeout.Infinite);

Run dotnet run worker.cs in one terminal and dotnet run publisher.cs in another.

A runnable example is in dotnet/mqttnet in cloudamqp/examples.

Good to know

Put the vhost in the username, as vhost:user. MQTT has no vhost of its own, and the console shows the username that way.
Connect with MQTT 3.1.1. MQTTnet 5 defaults to MQTT 5, which LavinMQ doesn’t support, so set WithProtocolVersion(MqttProtocolVersion.V311).
UseTls() verifies the server certificate.
Use QoS 1, so PublishStringAsync completes once the broker has sent its PUBACK, and each message is acked when the handler’s task completes.
MQTTnet 5 doesn’t reconnect. Handle DisconnectedAsync, connect again and subscribe again.

← GO

NodeMCU →