C# Demo

Prerequisites

Knowledge of basic C# syntax and how to configure .NET Framework development environments.

Development Environment

In this example, .NET Framework 4.6.2 and .NET SDK 6.0.421 are used. Download them from the .NET official website. After installation, run the following command to check the version:

dotnet -v

Dependency

In this example, MQTTnet and MQTTnet.Extension.ManagedClient (version 3.0.11) are used. You can search for MQTTnet in the NuGet manager and install the required version.

Figure 1 nuget installation dependency

Sample Code

ClientConf.cs code:

using MQTTnet.Protocol;

namespace mqttcs
{
    public class ClientConf
    {
        // MQTT subscription address
        public string ServerUri { get; set; }

        // MQTT subscription port number
        public int Port { get; set; }
        
        // MQTT access credential access_key
        public string AccessKey { get; set; }
        
        // MQTT access credential access_code
        public string AccessCode { get; set; }
        
        // MQTT client ID
        public string ClientId { get; set; }
        
        // Instance ID. This parameter is mandatory when multiple instances of the standard edition are purchased in the same region.
        public string InstanceId { get; set; }
        
        // MQTT subscription topic
        public string Topic { get; set; }
        
        // mqtt qos
        public MqttQualityOfServiceLevel Qos { get; set; }
        
    }
}

MqttListener code:

using System;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Extensions.ManagedClient;

namespace mqttcs
{
    public interface MqttListener
    {
        // Callback function when the MQTT client is disconnected from the server
        void ConnectionLost(MqttClientDisconnectedEventArgs e);

        // Callback function for successful connection establishment between the MQTT client and server
        void ConnectComplete(MqttClientConnectResultCode resultCode, String reason);
        
        // Callback function for consuming messages on the MQTT client
        void OnMessageReceived(String message);
        
        // Callback function when the MQTT client fails to establish a connection with the server
        void ConnectFail(ManagedProcessFailedEventArgs e);
    }
}

MqttConnection.cs code:

using System;
using System.Text;
using System.Threading;
using MQTTnet;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Options;
using MQTTnet.Client.Receiving;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Formatter;

namespace mqttcs
{
    public class MqttConnection
    {
        private static IManagedMqttClient client = null;
        
        private static ManualResetEvent mre = new ManualResetEvent(false);
        
        private static readonly ushort DefaultKeepLive = 120;
        
        private static int _retryTimes = 0;
        
        private readonly int _retryTimeWait = 1000;
        
        private readonly ClientConf _clientConf;

        private MqttListener _listener;
        
        public MqttConnection(ClientConf clientConf, MqttListener listener)
        {
            _clientConf = clientConf;
            _listener = listener;
        }
        
        public int Connect()
        {
            client?.StopAsync();
        // Backoff retry from 1s to 20s
            var duration = 1000;
            var maxDuration = 20 * 1000;
            var rc = InternalConnect();
            while (rc != 0)
            {
                Thread.Sleep((int)duration);
                if (duration < maxDuration)
                {
                    duration *= 2;
                }
                client?.StopAsync();
                _retryTimes++;
                Console.WriteLine("connect mqtt broker retry. times: " + _retryTimes);
                rc = InternalConnect();
            }

            return rc;
        }

        private int InternalConnect()
        {
            try
            {
                client = new MqttFactory().CreateManagedMqttClient();
                client.ApplicationMessageReceivedHandler =
                    new MqttApplicationMessageReceivedHandlerDelegate(ApplicationMessageReceiveHandlerMethod);
                client.ConnectedHandler = new MqttClientConnectedHandlerDelegate(OnMqttClientConnected);
                client.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(OnMqttClientDisconnected);
                client.ConnectingFailedHandler = new ConnectingFailedHandlerDelegate(OnMqttClientConnectingFailed);
                IManagedMqttClientOptions options = GetOptions();
                // Connects to the platform.
                client.StartAsync(options);
                mre.Reset();

                mre.WaitOne();
                if (!client.IsConnected)
                {
                    return -1;
                }

                var mqttTopicFilter = new MqttTopicFilterBuilder().WithTopic(_clientConf.Topic).WithQualityOfServiceLevel(_clientConf.Qos).Build();
                
                client.SubscribeAsync(mqttTopicFilter).Wait();
                Console.WriteLine("subscribe topic success.");
                return 0;
            }
            catch (Exception e)
            {
                Console.WriteLine("Connect to mqtt server failed. err: " + e);
                return -1;
            }
        }

        private void ApplicationMessageReceiveHandlerMethod(MqttApplicationMessageReceivedEventArgs e)
        {
            string payload = null;
            if (e.ApplicationMessage.Payload != null)
            {
                payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
            }
            try
            {
                _listener?.OnMessageReceived(payload);
            }
            catch (Exception ex)
            {
                Console.WriteLine("Message received error, the message is " + payload);
            }
            
        }
        
        private void OnMqttClientConnected(MqttClientConnectedEventArgs e)
        {
            try
            {
                _retryTimes = 0;
                _listener?.ConnectComplete(e.AuthenticateResult.ResultCode, e.AuthenticateResult.ReasonString);
                mre.Set();
            }
            catch (Exception exception)
            {
                Console.WriteLine("handle connect callback failed. e: " + exception.Message);
            }
        }
        
        private void OnMqttClientDisconnected(MqttClientDisconnectedEventArgs e)
        {
            try
            {
                _listener?.ConnectionLost(e);
            }
            catch (Exception exception)
            {
                Console.WriteLine("handle disConnect callback failed. e: " + exception.Message);
            }
            
        }

        private void OnMqttClientConnectingFailed(ManagedProcessFailedEventArgs e)
        {
            try
            {
                if (_listener != null)
                {
                    _listener.ConnectFail(e);
                }
                Thread.Sleep(_retryTimeWait);
                Connect();
            }
            catch (Exception exception)
            {
                Console.WriteLine("handle connect failed callback failed. e: " + exception.Message);
            }
        }

        private IManagedMqttClientOptions GetOptions()
        {
            IManagedMqttClientOptions options = null;
            long timestamp = new DateTimeOffset(DateTime.UtcNow).ToUnixTimeMilliseconds();
            string userName = "accessKey=" + _clientConf.AccessKey + "|timestamp=" + timestamp + "|instanceId=" + _clientConf.InstanceId;

            options = new ManagedMqttClientOptionsBuilder()
                .WithClientOptions(new MqttClientOptionsBuilder()
                    .WithTcpServer(_clientConf.ServerUri, _clientConf.Port)
                    .WithCredentials(userName, _clientConf.AccessCode)
                    .WithClientId(_clientConf.ClientId)
                    .WithKeepAlivePeriod(TimeSpan.FromSeconds(DefaultKeepLive))
                    .WithTls(new MqttClientOptionsBuilderTlsParameters()
                    {
                        AllowUntrustedCertificates = true,
                        UseTls = true,
                        CertificateValidationHandler = delegate { return true; },
                        IgnoreCertificateChainErrors = false,
                        IgnoreCertificateRevocationErrors = false,
                        SslProtocol = System.Security.Authentication.SslProtocols.Tls12,
                    })
                    .WithProtocolVersion(MqttProtocolVersion.V500)
                    .Build())
                .Build();
            return options;
        }
    }
}

MqttClient.cs code:

using System;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Protocol;

namespace mqttcs
{
    class MqttClient: MqttListener
    {
        private static ManualResetEvent mre = new ManualResetEvent(false);

        public static async Task Main(string[] args)
        {
            ClientConf clientConf = new ClientConf();
            clientConf.ClientId = "your mqtt clientId";
            clientConf.ServerUri = "your mqtt host";
            clientConf.Port = 8883;
            clientConf.AccessKey = Environment.GetEnvironmentVariable("MQTT_ACCESS_KEY");
            clientConf.AccessCode = Environment.GetEnvironmentVariable("MQTT_ACCESS_CODE");
            clientConf.InstanceId = "your instanceId";
            clientConf.Topic = "your mqtt topic";
            clientConf.Qos = MqttQualityOfServiceLevel.AtMostOnce;

            MqttConnection connection = new MqttConnection(clientConf, new MqttClient());
            var connect = connection.Connect();
            if (connect == 0)
            {
                Console.WriteLine("success to init mqtt connection.");
                mre.WaitOne();
            }
        }

        public void ConnectionLost(MqttClientDisconnectedEventArgs e)
        {
            if (e?.Exception != null)
            {
                Console.WriteLine("connect was lost. exception: " + e.Exception.Message);
                return;
            }
            Console.WriteLine("connect was lost");
            
        }

        public void ConnectComplete(MqttClientConnectResultCode resultCode, String reason)
        {
            Console.WriteLine("connect success. resultCode: " + resultCode + " reason: " + reason);
        }

        public void OnMessageReceived(string message)
        {
            Console.WriteLine("receive msg: " + message);
        }

        public void ConnectFail(ManagedProcessFailedEventArgs e)
        {
            Console.WriteLine("connect mqtt broker failed. e: " + e.Exception.Message);
        }
    }
}

Success Example

After the access is successful, the following information is displayed on the client.

Figure 2 Example of successful client access using C#