zengzy / hhrcs_service · Files

using System.Collections.Concurrent;
using System.Text.Json;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using MQTTnet.Client;
using Rcs.Application.Services;
using Rcs.Application.Services.PathFind;
using Rcs.Application.Services.PathFind.Models;
using Rcs.Application.Services.Protocol;
using Rcs.Application.Shared;
using Rcs.Domain.Entities;
using Rcs.Domain.Models.VDA5050;
using Rcs.Domain.Settings;
using Rcs.Infrastructure.Mqtt.ParseFactory;
using Rcs.Infrastructure.PathFinding.Services;
using Rcs.Infrastructure.Services.Protocol;
using StackExchange.Redis;
using Task = System.Threading.Tasks.Task;
using TaskStatus = System.Threading.Tasks.TaskStatus;

namespace Rcs.Infrastructure.Mqtt
{
    /// <summary>
    /// MQTT消息处理器，支持动态注册和动态字符串
    /// </summary>
    public class MqttMessageHandler : IMqttMessageHandler
    {
        private readonly ILogger<MqttMessageHandler> _logger;
        private readonly IServiceProvider _serviceProvider;
        private readonly IStateParserFactory _stateParserFactory;
        private readonly IVisualizationParserFactory _visualizationParserFactory;
        private readonly IRobotCacheService _robotCacheService;
        private readonly IUnifiedTrafficControlService _unifiedTrafficControlService;
        private readonly ConcurrentDictionary<string, Func<string, string, string, Task<MqttHandleResult>>> _handlers;
        private readonly ConcurrentDictionary<string, int> _topicHeaderId = new()
        {
            ["connection"] = 0,
            ["state"] = 0,
            ["factsheet"] = 0,
            ["visualization"] = 0
        };
        private DateTime _lastLogTime = DateTime.MinValue;
        private readonly IConnectionMultiplexer _redis;
        private readonly AppSettings _settings;
        private readonly TimeSpan _logInterval = TimeSpan.FromSeconds(15);

        /// <summary>
        /// 使用延迟解析获取协议服务工厂，避免循环依赖：
        /// Vda5050ProtocolService → IMqttClientService → IMqttMessageHandler → IProtocolServiceFactory
        /// </summary>
        private IProtocolServiceFactory ProtocolServiceFactory => _serviceProvider.GetRequiredService<IProtocolServiceFactory>();

        public MqttMessageHandler(
            ILogger<MqttMessageHandler> logger,
            IServiceProvider serviceProvider,
            IStateParserFactory stateParserFactory,
            IVisualizationParserFactory visualizationParserFactory,
            IRobotCacheService robotCacheService,
            IUnifiedTrafficControlService unifiedTrafficControlService,
            IConnectionMultiplexer redis,
            IOptions<AppSettings> settings)
        {
            _logger = logger;
            _serviceProvider = serviceProvider;
            _stateParserFactory = stateParserFactory;
            _visualizationParserFactory = visualizationParserFactory;
            _robotCacheService = robotCacheService;
            _unifiedTrafficControlService = unifiedTrafficControlService;
            _handlers = new ConcurrentDictionary<string, Func<string, string, string, Task<MqttHandleResult>>>();
            _redis = redis;
            _settings = settings.Value;
            // 注册默认处理器
            RegisterDefaultHandlers();
        }

        /// <summary>
        /// 处理接收到的MQTT消息
        /// </summary>
        public async Task HandleMessageAsync(MqttApplicationMessageReceivedEventArgs e)
        {
            try
            {
                var topic = e.ApplicationMessage.Topic;
                var payload = ExtractPayload(e.ApplicationMessage);
                if (!topic.Contains("visualization"))
                {
                    var now = DateTime.Now;
                    if (now - _lastLogTime >= _logInterval)
                    {
                        _logger.LogInformation("[MQTT] Recive | Topic={Topic} | Payload={Payload}", topic, payload);
                        _lastLogTime = now;
                    }
                }

                var result = await ProcessMessageAsync(topic, payload);

                if (!result.Success)
                {
                    _logger.LogWarning("[Advanced MQTT Handler] 消息处理失败 | Topic={Topic} | Error={Error}",
                        topic, result.Message);
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "[Advanced MQTT Handler] 消息处理过程中发生严重错误");
            }
        }

        /// <summary>
        /// 注册消息类型处理器
        /// </summary>
        /// <param name="messageType">消息类型</param>
        /// <param name="handler">处理器函数</param>
        public void RegisterHandler(string messageType, Func<string, string, string, Task<MqttHandleResult>> handler)
        {
            _handlers.AddOrUpdate(messageType.ToLowerInvariant(), handler, (key, oldValue) => handler);
            _logger.LogInformation("[Advanced MQTT Handler] 注册处理器 | MessageType={MessageType}", messageType);
        }

        /// <summary>
        /// 处理消息
        /// </summary>
        private async Task<MqttHandleResult> ProcessMessageAsync(string topic, string payload)
        {
            var topicParts = topic.Split('/');

            if (topicParts.Length < 5)
            {
                return MqttHandleResult.CreateFailure($"主题格式不正确: {topic}");
            }

            var robotSerialNumber = topicParts[3];
            var messageType = topicParts[4].ToLowerInvariant();
            // robot制造商
            var robotManufacturer = topicParts[2];

            if (_handlers.TryGetValue(messageType, out var handler))
            {
                return await handler(robotManufacturer,robotSerialNumber, payload);
            }
            else
            {
                return MqttHandleResult.CreateFailure($"未找到消息类型 '{messageType}' 的处理器");
            }
        }

        /// <summary>
        /// 注册默认处理器
        /// </summary>
        private void RegisterDefaultHandlers()
        {
            RegisterHandler("connection", HandleConnectionMessageAsync);
            RegisterHandler("state", HandleStateMessageAsync);
            RegisterHandler("factsheet", HandleFactsheetMessageAsync);
            RegisterHandler("visualization", HandleVisualizationMessageAsync);
        }

        /// <summary>
        /// 处理连接状态消息
        /// </summary>
        private async Task<MqttHandleResult> HandleConnectionMessageAsync(string robotManufacturer, string robotSerialNumber, string payload)
        {
            try
            {
                var connectionInfo = JsonSerializer.Deserialize<Connection>(payload);
                if (connectionInfo == null) throw new Exception();
                // 重置该机器人所有topic的headerId为0
                if (!string.IsNullOrEmpty(robotManufacturer) && !string.IsNullOrEmpty(robotSerialNumber))
                {
                    await ResetHeaderIdsAsync(robotManufacturer,robotSerialNumber);
                }
                var robotStatus = connectionInfo.ConnectionState switch
                {
                    "ONLINE" => OnlineStatus.Online,
                    "OFFLINE" => OnlineStatus.Offline,
                    "CONNECTIONBROKEN" => OnlineStatus.Connectionbroken,
                    _ => OnlineStatus.Offline
                };

                return MqttHandleResult.CreateSuccess($"更新机器人连接状态: {robotSerialNumber} -> {connectionInfo.ConnectionState}");

            }
            catch (Exception ex)
            {
                return MqttHandleResult.CreateFailure($"处理MQTT连接消息失败: {ex.Message}", ex);
            }
        }

        /// <summary>
        /// 处理状态消息（直接使用制造商+序列号更新缓存）
        /// @author zzy
        /// </summary>
        private async Task<MqttHandleResult> HandleStateMessageAsync(string robotManufacturer, string robotSerialNumber, string payload)
        {
            try
            {

                // 使用工厂解析特定制造商的状态数据
                var stateInfo = _stateParserFactory.ParseState(robotManufacturer, payload);
                if (stateInfo == null) throw new Exception("状态数据解析失败");
                // 可选：进行headerId去重判断
                if (!await IsNewHeaderIdAsync("state", robotManufacturer, robotSerialNumber, stateInfo.HeaderId))
                    throw new Exception($"状态消息HeaderId: {stateInfo.HeaderId} 消息过期");
                // 解析状态
                var robotStatus = DetermineRobotStatus(stateInfo);
                var operatingMode = ParseOperatingMode(stateInfo.OperatingMode);
                var errors = stateInfo.Errors.Any() ? JsonSerializer.Serialize(stateInfo.Errors) : null;

                // 直接使用制造商+序列号更新状态到Redis缓存
                await _robotCacheService.UpdateStatusAsync(
                    robotManufacturer,
                    robotSerialNumber,
                    robotStatus,
                    OnlineStatus.Online,
                    (int?)stateInfo.BatteryState?.BatteryCharge,
                    stateInfo.Driving,
                    stateInfo.Paused,
                    stateInfo.BatteryState?.Charging,
                    operatingMode,
                    errors);

                // 更新位置到Redis缓存
                if (stateInfo.AgvPosition != null)
                {
                    await _robotCacheService.UpdateLocationAsync(
                        robotManufacturer,
                        robotSerialNumber,
                        null,
                        null,
                        stateInfo.AgvPosition.X,
                        stateInfo.AgvPosition.Y,
                        stateInfo.AgvPosition.Theta);
                }

                // 机器人行驶中时，续期路径锁（防止30秒TTL过期导致冲突误判）
                // @author zzy
                // @date 2026-01-30
                if (stateInfo.Driving)
                {
                    var basicInfo = await _robotCacheService.GetBasicAsync(robotManufacturer, robotSerialNumber);
                    if (basicInfo != null && Guid.TryParse(basicInfo.RobotId, out var robotId))
                    {
                        await _unifiedTrafficControlService.RenewAllRobotLocksAsync(robotId, 30);
                    }
                }

                // 处理NewBaseRequest：当车辆快到当前段终点时，判断并发送下一段路径指令
                if ((bool)stateInfo.NewBaseRequest)
                {
                    await HandleNewBaseRequestAsync(robotManufacturer, robotSerialNumber);
                }
                return MqttHandleResult.CreateSuccess($"成功更新机器人状态: {robotManufacturer}:{robotSerialNumber}");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "处理制造商 '{Manufacturer}' 的状态消息失败", robotManufacturer);
                return MqttHandleResult.CreateFailure($"处理状态消息失败: {ex.Message}", ex);
            }
        }

        /// <summary>
        /// 处理设备信息消息
        /// </summary>
        private Task<MqttHandleResult> HandleFactsheetMessageAsync(string robotManufacturer, string robotSerialNumber, string payload)
        {
            try
            {
                //var factSheet = JsonSerializer.Deserialize<FactSheet>(payload);
                //if (factSheet == null) throw new SharedException(10001);
                //if (!IsNewHeaderId("factsheet", factSheet.HeaderId)) throw new SharedException(10002);

                //using var scope = _serviceProvider.CreateScope();
                //var robotRepository = scope.ServiceProvider.GetRequiredService<IRobotRepository>();

                //var robot = await robotRepository.GetBySerialNumberAsync(robotSerialNumber);
                //if (robot == null)
                //{
                //    throw new ExtendRobotException(robotSerialNumber, 20003);
                //}
                //robot.UpdateFactSheet(factSheet);
                //await robotRepository.UpdateAsync(robot);

                return Task.FromResult(MqttHandleResult.CreateSuccess());
            }
            catch (Exception ex)
            {
                return Task.FromResult(MqttHandleResult.CreateFailure($"处理设备信息失败: {ex.Message}", ex));
            }
        }

        /// <summary>
        /// 处理可视化消息
        /// </summary>
        private async Task<MqttHandleResult> HandleVisualizationMessageAsync(string robotManufacturer, string robotSerialNumber, string payload)
        {
            try
            {
                var visualization = _visualizationParserFactory.ParseVisualization(robotManufacturer, payload);
                if (visualization == null) return MqttHandleResult.CreateFailure("可视化数据解析失败");

                // 可选：进行headerId去重判断
                if (!await IsNewHeaderIdAsync("visualization", robotManufacturer, robotSerialNumber, visualization.HeaderId))
                    throw new Exception($"可视化消息HeaderId: {visualization.HeaderId} 消息过期");
                
                try
                {
                    
                    // var redisDb = _redisHelper.GetDatabaseWithPrefix();
                    // var robot = await redisDb.StringGetAsync($"robot:{robotSerialNumber}");
                    // if (robot.HasValue) {
                    //     var robotPosition = JsonSerializer.Deserialize<Robot>(robot, new JsonSerializerOptions
                    //     {
                    //         PropertyNamingPolicy = JsonNamingPolicy.CamelCase
                    //     });
                    //     if (visualization is Visualization_HikRobot visualization_HikRobot) {
                    //        robotOld.CurrentX = visualization_HikRobot.AgvPosition.X * robotOld.CoordinateScale;
                    //        robotOld.CurrentY = visualization_HikRobot.AgvPosition.Y * robotOld.CoordinateScale;
                    //        robotOld.CurrentTheta = visualization_HikRobot.AgvPosition.Theta;
                    //     }
                    //     var robotNew = JsonSerializer.Serialize(robotOld, new JsonSerializerOptions
                    //     {
                    //         PropertyNamingPolicy = JsonNamingPolicy.CamelCase
                    //     });
                    //     await redisDb.StringSetAsync($"robot:{robotSerialNumber}", robotNew);
                    // }
                    
                    //_logger.LogDebug("成功将机器人 {RobotCode} 数据存储到Redis", robot.RobotCode);
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, "存储机器人 {RobotCode} 数据到Redis失败", robotSerialNumber);
                }
                
                // 目前仅确认解析成功即可
                return MqttHandleResult.CreateSuccess();
            }
            catch (Exception ex)
            {
                return MqttHandleResult.CreateFailure($"处理可视化消息失败: {ex.Message}", ex);
            }
        }

        /// <summary>
        /// 从MQTT消息中提取载荷
        /// </summary>
        private static string ExtractPayload(MQTTnet.MqttApplicationMessage message)
        {
            return message.PayloadSegment.Array is { } arr
                ? System.Text.Encoding.UTF8.GetString(arr, message.PayloadSegment.Offset, message.PayloadSegment.Count)
                : string.Empty;
        }

        private string BuildHeaderHashKey(string robotManufacturer, string robotSN)
        {
            return $"{_settings.Redis.KeyPrefixes.Robot}:{robotManufacturer}:{robotSN}:{_settings.Redis.KeyPrefixes.MqttHeaderSuffix}";
        }

        // 基于Redis按机器人维度检查并更新headerId
        private async Task<bool> IsNewHeaderIdAsync(string topic, string robotManufacturer,string robotSN, int newHeaderId)
        {
            try
            {
                var redisDb = _redis.GetDatabase();
                var key = BuildHeaderHashKey(robotManufacturer,robotSN);
                var field = topic.ToLowerInvariant();

                var currentValue = await redisDb.HashGetAsync(key, field);
                var currentHeaderId = 0;
                if (currentValue.HasValue && int.TryParse(currentValue.ToString(), out var parsed))
                {
                    currentHeaderId = parsed;
                }
                
                if (newHeaderId > currentHeaderId)
                {
                    await redisDb.HashSetAsync(key, field, newHeaderId);
                    return true;
                }

                return false;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "检查并更新机器人 {RobotCode} 的 {Topic} headerId 失败", robotManufacturer + robotSN, topic);
                return false;
            }
        }

        // 将指定机器人的所有topic headerId重置为0
        public async Task ResetHeaderIdsAsync( string robotManufacturer,string robotSN)
        {
            try
            {
                var redisDb = _redis.GetDatabase();
                var key = BuildHeaderHashKey(robotManufacturer,robotSN);
                var entries = new HashEntry[]
                {
                    new("connection", 0),
                    new("state", 0),
                    new("factsheet", 0),
                    new("visualization", 0)
                };
                await redisDb.HashSetAsync(key, entries);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "重置机器人 {RobotCode} 的 headerId 失败", robotManufacturer + robotSN );
            }
        }      
        /// <summary>
        /// 根据状态信息判断Robot状态
        /// </summary>
        private RobotStatus DetermineRobotStatus(State stateInfo)
        {
            // 判断是否有致命错误
            var hasFatalError = stateInfo.Errors.Any(e =>
                e.ErrorLevel?.Equals("FATAL", StringComparison.OrdinalIgnoreCase) == true);

            if (hasFatalError)
                return RobotStatus.Error;

            // 判断是否忙碌（正在行驶或暂停）
            if (stateInfo.Driving || stateInfo.Paused == true)
                return RobotStatus.Busy;

            // 默认空闲
            return RobotStatus.Idle;
        }

        /// <summary>
        /// 解析操作模式
        /// @author zzy
        /// </summary>
        private static OperatingMode ParseOperatingMode(string? mode)
        {
            return mode?.ToUpperInvariant() switch
            {
                "AUTOMATIC" => OperatingMode.Automatic,
                "SEMIAUTOMATIC" => OperatingMode.Semiautomatic,
                "MANUAL" => OperatingMode.Manual,
                "SERVICE" => OperatingMode.Service,
                "TEACHIN" => OperatingMode.Teachin,
                _ => OperatingMode.Automatic
            };
        }

        /// <summary>
        /// 处理NewBaseRequest请求，当车辆快到当前段终点时，调用任务服务发送下一段路径指令
        /// @author zzy
        /// </summary>
        /// <param name="robotManufacturer">机器人制造商</param>
        /// <param name="robotSerialNumber">机器人序列号</param>
        /// <summary>
        /// 处理NewBaseRequest请求，当车辆快到当前段终点时，调用任务服务发送下一段路径指令
        /// @author zzy
        /// </summary>
        /// <param name="robotManufacturer">机器人制造商</param>
        /// <param name="robotSerialNumber">机器人序列号</param>
        private async Task HandleNewBaseRequestAsync(string robotManufacturer, string robotSerialNumber)
        {
            try
            {
                // 使用延迟解析获取协议服务，避免循环依赖
                var protocol = ProtocolServiceFactory.GetService(ProtocolType.VDA);

                // 调用任务服务发送下一段路径指令
                await protocol.SendNextSegmentAsync(robotManufacturer, robotSerialNumber);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "处理NewBaseRequest失败: Robot={Manufacturer}:{SerialNumber}",
                    robotManufacturer, robotSerialNumber);
            }
        }
    }
}