IotCloudPlatform/IoTIntegrationPlatform/IoTIntegrationPlatform.Common/MqttHelper.cs


using System;
using System.Threading;
using System.Threading.Tasks;
using static IoTIntegrationPlatform.Model.Enum.EnumMqtt;
using IoTIntegrationPlatform.Model.Model;
using IoTIntegrationPlatform.Model.MqttModel;

namespace IoTIntegrationPlatform.Common
{
    /// <summary>
    /// mqtt公共类
    /// </summary>
    public static class MqttHelper
    {
        /// <summary>
        /// 获取设备网关编号
        /// </summary>
        /// <param name="topic"></param>
        /// <returns></returns>
        public static string GetDeviceMac(string topic)
        {
            string deviceMac = string.Empty;
            try
            {
                if (!string.IsNullOrEmpty(topic))
                {
                    var topicArray = topic.Split('/');
                    if (topicArray != null && topicArray.Length > 2)
                    {
                        deviceMac = topicArray[2].ToString();//网关id
                    }
                }
            }
            catch (Exception ex)
            {
                Logging.Error(ex, ex.Message);
            }
            return deviceMac;
        }

        /// <summary>
        /// 获取mqttTopic类型
        /// </summary>
        /// <param name="topic"></param>
        /// <returns></returns>
        public static EnumMqttTopicType GetMqttTopicType(string topic)
        {
            EnumMqttTopicType topicType = EnumMqttTopicType.client_connect;
            try
            {
                var topicArray = topic.Split('/');
                var topicName = topicArray[topicArray.Length - 1];//topic名称
                if (!string.IsNullOrEmpty(topic))
                {
                    Enum.TryParse(topicName, out topicType);
                }
            }
            catch (Exception ex)
            {
                Logging.Error(ex, ex.Message);
            }
            return topicType;
        }


       


        /// <summary>
        /// 获取任务执行结果
        /// </summary>
        /// <param name="taskId"></param>
        /// <returns></returns>
        public static bool GetTaskExecutionResult(string taskId)
        {
            bool executionResult = false;
            try
            {
                //最大秒数
                taskId = RedisKeyHelper.MqttTask + "_" + taskId;
                int maxSeconds = 5;
                int totalCount = maxSeconds * 20;
                for (int i = 0; i < totalCount; i++)
                {
                    Thread.Sleep(50);
                    MqttTaskModel taskModel = RedisHelper.Instance.StringGet<MqttTaskModel>(taskId);
                    if (taskModel != null && taskModel.Status == 2)
                    {
                        return true;
                    }
                }
            }
            catch (Exception ex)
            {
                Logging.Error(ex, ex.Message);
            }
            return executionResult;

        }

        /// <summary>
        /// 执行超时任务
        /// </summary>
        /// <param name="conditionFunc"></param>
        /// <param name="timeout"></param>
        /// <returns></returns>
        public static async Task PerformTaskWithTimeout(Func<bool> conditionFunc, TimeSpan timeout)
        {
            var tokenSource = new CancellationTokenSource();
            var token = tokenSource.Token;

            try
            {
                while (!conditionFunc() && !token.IsCancellationRequested)
                {
                    await Task.Delay(1000, token); // 等待1秒，然后再次检查条件，如果取消任务则跳出循环  
                }
            }
            finally
            {
                tokenSource.Cancel(); // 无论条件是否满足，都会取消任务  
            }
        }




    }
}