MQTT Publishing and Subscribing Messages to MQTT Broker (CloudMQTT) Using .NET C# MQTT Client Library

Recently, I was evaluating a few .NET C# MQTT Client Libraries. After a bit of research, I found the following interesting .NET C# MQTT Client Libraries:

1. MqttDotNet
2. paho.mqtt.m2mqtt
3. MQTTnet

After evaluating these, I found MQTTnet was the one that covers all my use cases. In this article, I will share how we can use MQTTnet .NET C# MQTT Client Library to publish and subscribe messages to MQTT Broker. I will be using CloudMQTT MQTT Broker Free Instance for this article.

Add "MQTTnet.Extensions.ManagedClient" Nuget Package

Benefits of Using Managed Client

• Auto-reconnect
• Internal queuing support
• Storage support that enables sending the messages even after application restart
• No need to subscribe manually after disconnection from the server

Connect to the MQTT Broker (CloudMQTT)

/// <summary>
/// Connect to broker.
/// </summary>
/// <returns>Task.</returns>
public static async Task ConnectAsync()
{
  string clientId = Guid.NewGuid().ToString();
  string mqttURI = {REPLACE THIS WITH YOUR MQTT SERVER URI HERE}
  string mqttUser = { REPLACE THIS WITH YOUR MQTT USER HERE }
  string mqttPassword = { REPLACE THIS WITH YOUR MQTT PASSWORD HERE }
  int mqttPort = { REPLACE THIS WITH YOUR MQTT PORT HERE }
  bool mqttSecure = {IF YOU ARE USING SSL Port THEN SET true OTHERWISE SET false}

  var messageBuilder = new MqttClientOptionsBuilder()
    .WithClientId(clientId)
    .WithCredentials(mqttUser, mqttPassword)
    .WithTcpServer(mqttURI, mqttPort)
    .WithCleanSession();

  var options = mqttSecure
    ? messageBuilder
      .WithTls()
      .Build()
    : messageBuilder
      .Build();

  var managedOptions = new ManagedMqttClientOptionsBuilder()
    .WithAutoReconnectDelay(TimeSpan.FromSeconds(5))
    .WithClientOptions(options)
    .Build();

  client = new MqttFactory().CreateManagedMqttClient();

  await client.StartAsync(managedOptions);
}

Set Up the CloudMQTT Broker Instance and Get the Instance Info Required for Connection

Create New Instance

Select a Plan and Name

Select a Region and Data Center

Confirm New Instance

Select the Newly Created Instance

Get the Broker Instance Info

Publish Messages to the MQTT Broker (CloudMQTT)

/// <summary>
/// Publish Message.
/// </summary>
/// <param name="topic">Topic.</param>
/// <param name="payload">Payload.</param>
/// <param name="retainFlag">Retain flag.</param>
/// <param name="qos">Quality of Service.</param>
/// <returns>Task.</returns>
public static async Task PublishAsync(string topic, string payload, bool retainFlag = true, int qos = 1) => 
  await client.PublishAsync(new MqttApplicationMessageBuilder()
    .WithTopic(topic)
    .WithPayload(payload)
    .WithQualityOfServiceLevel((MQTTnet.Protocol.MqttQualityOfServiceLevel)qos)
    .WithRetainFlag(retainFlag)
    .Build());

Here, Quality of Service (QoS) can be:

• At most once (0)

• At least once (1)

• Exactly once (2)

Retain Message Flag Can Be True/False

While publishing, we can tell the MQTT broker to keep the last message on that topic by setting the retained message flag to true.

Subscribe Messages From the MQTT Broker (CloudMQTT)

/// <summary>
/// Subscribe topic.
/// </summary>
/// <param name="topic">Topic.</param>
/// <param name="qos">Quality of Service.</param>
/// <returns>Task.</returns>
public static async Task SubscribeAsync(string topic, int qos = 1) =>
  await client.SubscribeAsync(new TopicFilterBuilder()
    .WithTopic(topic)
    .WithQualityOfServiceLevel((MQTTnet.Protocol.MqttQualityOfServiceLevel)qos)
    .Build());

Handle Events

Connected Handler

client.UseConnectedHandler(e =>
{
  Console.WriteLine("Connected successfully with MQTT Brokers.");
});

Disconnected Handler

client.UseDisconnectedHandler(e =>
{
  Console.WriteLine("Disconnected from MQTT Brokers.");
});

Message Received Handler

You can use this for processing the subscribed messages.

client.UseApplicationMessageReceivedHandler(e =>
{
  try
  {
    string topic = e.ApplicationMessage.Topic;

    if (string.IsNullOrWhiteSpace(topic) == false)
    {
      string payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
      Console.WriteLine($"Topic: {topic}. Message Received: {payload}");
    }
  }
  catch (Exception ex)
  {
    Console.WriteLine(ex.Message, ex);
  }
});

Check Published Messages on MQTT Broker (CloudMQTT)

That's all for now!