1440WGJ/JY.Inspection/Common/MqttSingleton.cs

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

namespace JY.Inspection.Common
{
    public class MqttSingleton
    {
        // 单例实例
        private static readonly Lazy<MqttSingleton> _instance =
            new Lazy<MqttSingleton>(() => new MqttSingleton());

        public static MqttSingleton Instance => _instance.Value;

        // 内部成员
        private IMqttClient _client;
        private MqttClientOptions _options;
        private bool _isInitialized = false;

        // 事件：收到消息时触发
        public event Action<string, string> MessageReceived;

        // 私有构造函数
        private MqttSingleton() { }

        /// <summary>
        /// 初始化 MQTT 客户端（只需调用一次）
        /// </summary>
        /// <param name="brokerHost">MQTT 服务端地址</param>
        /// <param name="port">端口，默认1883</param>
        /// <param name="clientId">客户端ID，为空则自动生成</param>
        /// <param name="username">用户名（可选）</param>
        /// <param name="password">密码（可选）</param>
        /// <param name="useTls">是否启用TLS加密</param>
        public async Task InitializeAsync(
            string brokerHost,
            int port = 1883,
            string clientId = null,
            string username = null,
            string password = null,
            bool useTls = false)
        {
            if (_isInitialized) return;

            // 创建客户端
            var factory = new MqttFactory();
            _client = factory.CreateMqttClient();

            // 构建连接选项
            var optionsBuilder = new MqttClientOptionsBuilder()
                .WithTcpServer(brokerHost, port)
                .WithClientId(string.IsNullOrEmpty(clientId) ? $"WinForm_{Guid.NewGuid()}" : clientId)
                .WithCleanSession();

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

            if (useTls)
            {
                optionsBuilder.WithTlsOptions(new MqttClientTlsOptions() { 
                    UseTls = true
                });
            }

            _options = optionsBuilder.Build();

            // 绑定事件
            _client.ConnectedAsync += async e =>
            {
                Console.WriteLine("MQTT 已连接");
            };

            _client.DisconnectedAsync += async e =>
            {
                Console.WriteLine("MQTT 已断开，尝试重连...");
                // 自动重连
                await Task.Delay(3000).ContinueWith(_ => ConnectAsync());
            };

            _client.ApplicationMessageReceivedAsync += async e =>
            {
                var topic = e.ApplicationMessage.Topic;
                var payload = e.ApplicationMessage.PayloadSegment == null
                    ? null
                    : Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment.Array);

                MessageReceived?.Invoke(topic, payload);
            };

            // 连接
            await ConnectAsync();
            _isInitialized = true;
        }

        /// <summary>
        /// 连接MQTT服务端
        /// </summary>
        private async Task ConnectAsync()
        {
            try
            {
                if (!_client.IsConnected)
                {
                    await _client.ConnectAsync(_options, System.Threading.CancellationToken.None);
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine($"MQTT连接失败: {ex.Message}");
            }
        }

        /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="topic">主题</param>
        /// <param name="payload">消息内容</param>
        /// <param name="qos">QoS等级，默认1</param>
        public async Task SendAsync(string topic, string payload, int qos = 1)
        {
            if (!_client.IsConnected)
            {
                await ConnectAsync();
            }

            if (_client.IsConnected)
            {
                var message = new MqttApplicationMessageBuilder()
                    .WithTopic(topic)
                    .WithPayload(payload)
                    .WithQualityOfServiceLevel((MqttQualityOfServiceLevel)qos)
                    .WithRetainFlag(false)
                    .Build();

                await _client.PublishAsync(message, System.Threading.CancellationToken.None);
            }
        }

        /// <summary>
        /// 订阅主题
        /// </summary>
        public async Task SubscribeAsync(string topic)
        {
            if (_client.IsConnected)
            {
                await _client.SubscribeAsync(new MqttTopicFilterBuilder()
                    .WithTopic(topic)
                    .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
                    .Build());
            }
        }

        /// <summary>
        /// 断开连接
        /// </summary>
        public async Task DisconnectAsync()
        {
            if (_client != null && _client.IsConnected)
            {
                await _client.DisconnectAsync();
            }
        }

        /// <summary>
        /// 当前是否已连接
        /// </summary>
        public bool IsConnected => _client?.IsConnected ?? false;
    }
}