MqttManager.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using MQTTnet;


namespace dji_cloud_tool
{
    public class MqttManager
    {
        private IMqttClient? _mqttClient;
        private readonly MqttClientFactory _factory = new MqttClientFactory();

        public event Action<bool>? ConnectionStatusChanged;
        public event Action<string, string>? MessageReceived;

        public bool IsConnected => _mqttClient != null && _mqttClient.IsConnected;

        public async Task ConnectAsync(
            string host, 
            int port, 
            string username, 
            string password, 
            string clientId, 
            bool useTls, 
            bool ignoreCertErrors)
        {
            if (_mqttClient != null)
            {
                await DisconnectAsync();
            }

            _mqttClient = _factory.CreateMqttClient();

            var optionsBuilder = new MqttClientOptionsBuilder()
                .WithTcpServer(host, port)
                .WithClientId(clientId)
                .WithCleanSession(true);

            if (!string.IsNullOrWhiteSpace(username))
            {
                optionsBuilder.WithCredentials(username, password);
            }

            if (useTls)
            {
                optionsBuilder.WithTlsOptions(o =>
                {
                    o.UseTls();
                    if (ignoreCertErrors)
                    {
                        o.WithCertificateValidationHandler(_ => true);
                    }
                });
            }

            var options = optionsBuilder.Build();

            _mqttClient.ConnectedAsync += async e =>
            {
                await Task.CompletedTask;
                ConnectionStatusChanged?.Invoke(true);
            };

            _mqttClient.DisconnectedAsync += async e =>
            {
                await Task.CompletedTask;
                ConnectionStatusChanged?.Invoke(false);
            };

            _mqttClient.ApplicationMessageReceivedAsync += async e =>
            {
                await Task.CompletedTask;
                var payloadSeq = e.ApplicationMessage.Payload;
                string payload = "";
                if (payloadSeq.Length > 0)
                {
                    var bytes = System.Buffers.BuffersExtensions.ToArray(payloadSeq);
                    payload = Encoding.UTF8.GetString(bytes);
                }
                
                MessageReceived?.Invoke(e.ApplicationMessage.Topic, payload);
            };

            await _mqttClient.ConnectAsync(options);
        }

        public async Task DisconnectAsync()
        {
            if (_mqttClient == null) return;

            try
            {
                if (_mqttClient.IsConnected)
                {
                    await _mqttClient.DisconnectAsync();
                }
            }
            catch
            {
                // Ignore disconnect exceptions
            }
            finally
            {
                _mqttClient.Dispose();
                _mqttClient = null;
                ConnectionStatusChanged?.Invoke(false);
            }
        }

        public async Task SubscribeAsync(List<string> topics)
        {
            if (_mqttClient == null || !_mqttClient.IsConnected) return;

            var subscribeOptionsBuilder = new MqttClientSubscribeOptionsBuilder();
            
            foreach (var topic in topics)
            {
                subscribeOptionsBuilder.WithTopicFilter(f => f.WithTopic(topic));
            }

            await _mqttClient.SubscribeAsync(subscribeOptionsBuilder.Build());
        }

        public async Task UnsubscribeAsync(List<string> topics)
        {
            if (_mqttClient == null || !_mqttClient.IsConnected) return;

            var unsubscribeOptionsBuilder = new MqttClientUnsubscribeOptionsBuilder();

            foreach (var topic in topics)
            {
                unsubscribeOptionsBuilder.WithTopicFilter(topic);
            }

            await _mqttClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build());
        }

        public async Task PublishAsync(string topic, string payload)
        {
            if (_mqttClient == null || !_mqttClient.IsConnected) return;

            var message = new MqttApplicationMessageBuilder()
                .WithTopic(topic)
                .WithPayload(payload)
                .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce)
                .Build();

            await _mqttClient.PublishAsync(message);
        }
    }
}