Basic MQTT with C#

In this post, we're going to have a look on how to work with MQTT and Csharp. Creating a Broker and a Client that sends data to it.

But first...

What is MQTT?

According to the MQTT.org:

MQTT (short for Message Queuing Telemetry Transport) is an OASIS standard messaging protocol for the Internet of Things (IoT). It is designed as an extremely lightweight publish/subscribe messaging transport that is ideal for connecting remote devices with a small code footprint and minimal network bandwidth. MQTT today is used in a wide variety of industries, such as automotive, manufacturing, telecommunications, oil and gas, etc.

Coding

Project structure

The project structure contains a .sln file and two c# console projects, as it follows:

.
├── MQTTFirstLook.Broker
├── MQTTFirstLook.Client
└── MQTTFirstLook.sln

Libraries used

Broker implementation

This small broker implementation has the responsibility to read data received from connected clients and display it on the screen.

For this, we're going to create a new MQTTServer that listen to the port 707 on localhost.

Here is what is needed:

Usings

These are the libraries used by the server.

using System;
using System.Text;
using MQTTnet;
using MQTTnet.Server;
using Serilog;

Implementation

Here's how we create a new MQTTServer. In this example, this code is inside the void Main(string[] args) method.

// Create the options for our MQTT Broker
MqttServerOptionsBuilder options = new MqttServerOptionsBuilder()
                                     // set endpoint to localhost
                                     .WithDefaultEndpoint()
                                     // port used will be 707
                                     .WithDefaultEndpointPort(707) 
                                     // handler for new connections
                                     .WithConnectionValidator(OnNewConnection) 
                                     // handler for new messages
                                     .WithApplicationMessageInterceptor(OnNewMessage);

// creates a new mqtt server     
IMqttServer mqttServer = new MqttFactory().CreateMqttServer();

// start the server with options  
mqttServer.StartAsync(options.Build()).GetAwaiter().GetResult();

// keep application running until user press a key
Console.ReadLine();

Handlers

Handlers are callbacks that MQTT calls whenever an action is called. In this example, we have a handler for new connections and for whenever the server gets a new message.

public static void OnNewConnection(MqttConnectionValidatorContext context)
{
    Log.Logger.Information(
            "New connection: ClientId = {clientId}, Endpoint = {endpoint}",
            context.ClientId,
            context.Endpoint);
}

public static void OnNewMessage(MqttApplicationMessageInterceptorContext context)
{
    var payload = context.ApplicationMessage?.Payload == null ? null : Encoding.UTF8.GetString(context.ApplicationMessage?.Payload);

    MessageCounter++;

    Log.Logger.Information(
        "MessageId: {MessageCounter} - TimeStamp: {TimeStamp} -- Message: ClientId = {clientId}, Topic = {topic}, Payload = {payload}, QoS = {qos}, Retain-Flag = {retainFlag}",
        MessageCounter,
        DateTime.Now,
        context.ClientId,
        context.ApplicationMessage?.Topic,
        payload,
        context.ApplicationMessage?.QualityOfServiceLevel,
        context.ApplicationMessage?.Retain);
}

Client

In our client, we're going to create a new MQTTClient instance. This instance will connect to our Broker on localhost:707 and send messages to the topic Dev.to/topic/json

What is needed for this implementation is below:

Usings

These are the libraries used by the server.

using System;
using System.Threading.Tasks;
using MQTTnet;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Options;
using MQTTnet.Extensions.ManagedClient;
using Newtonsoft.Json;
using Serilog;

Implementation

Our MQTTClient will coonnect to our broker via TCP. And same as the server, this code will be inside void Main(string[] args) method.

// Creates a new client
MqttClientOptionsBuilder builder = new MqttClientOptionsBuilder()
                                        .WithClientId("Dev.To")
                                        .WithTcpServer("localhost", 707);

// Create client options objects
ManagedMqttClientOptions options = new ManagedMqttClientOptionsBuilder()
                        .WithAutoReconnectDelay(TimeSpan.FromSeconds(60))
                        .WithClientOptions(builder.Build())
                        .Build();

// Creates the client object
IManagedMqttClient _mqttClient = new MqttFactory().CreateManagedMqttClient();

// Set up handlers
_mqttClient.ConnectedHandler = new MqttClientConnectedHandlerDelegate(OnConnected);
_mqttClient.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(OnDisconnected);
_mqttClient.ConnectingFailedHandler = new ConnectingFailedHandlerDelegate(OnConnectingFailed);

// Starts a connection with the Broker
_mqttClient.StartAsync(options).GetAwaiter().GetResult();

// Send a new message to the broker every second
while (true)
{
    string json = JsonConvert.SerializeObject(new { message = "Heyo :)", sent= DateTimeOffset.UtcNow });
    _mqttClient.PublishAsync("dev.to/topic/json", json);

    Task.Delay(1000).GetAwaiter().GetResult();
}

Handlers

The client Handlers are called when the client recieves a signal that is connection, connection failed and when is disconnected.

public static void OnConnected(MqttClientConnectedEventArgs obj)
{
    Log.Logger.Information("Successfully connected.");
}

public static void OnConnectingFailed(ManagedProcessFailedEventArgs obj)
{
    Log.Logger.Warning("Couldn't connect to broker.");
}

public static void OnDisconnected(MqttClientDisconnectedEventArgs obj)
{
    Log.Logger.Information("Successfully disconnected.");
}