using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; using CoreAgent.ProtocolClient.HandlerEventArgs; using CoreAgent.ProtocolClient.Models; using Microsoft.Extensions.Logging; using Newtonsoft.Json.Linq; namespace CoreAgent.ProtocolClient.Managers { /// /// 消息ID管理器 /// public class MessageIdManager : IDisposable { #region 私有字段 private readonly ILogger _logger; private readonly string _clientName; private readonly ConcurrentDictionary _messageHandlers; private readonly ConcurrentDictionary _messageHandlersByName; // 使用long类型防止溢出 private long _generalMessageId; private long _logGetMessageId; // 状态管理 private bool _disposed; // 性能优化:缓存字符串构建器 private readonly StringBuilder _logBuilder = new StringBuilder(256); #endregion #region 事件 /// /// 日志获取ID变化事件 /// public event EventHandler? LogGetIdChanged; /// /// 消息处理器清理事件 /// public event EventHandler? HandlerCleanup; #endregion #region 属性 /// /// 当前通用消息ID /// public long CurrentGeneralMessageId => Interlocked.Read(ref _generalMessageId); /// /// 当前日志获取消息ID /// public long CurrentLogGetMessageId => Interlocked.Read(ref _logGetMessageId); /// /// 是否已释放 /// public bool IsDisposed => _disposed; /// /// 消息处理器数量 /// public int MessageHandlerCount => _messageHandlers.Count; /// /// 按名称的消息处理器数量 /// public int NamedMessageHandlerCount => _messageHandlersByName.Count; /// /// 总处理器数量 /// public int TotalHandlerCount => MessageHandlerCount + NamedMessageHandlerCount; #endregion #region 构造函数 /// /// 构造函数 /// public MessageIdManager(string clientName, ILogger logger) { _clientName = clientName ?? throw new ArgumentNullException(nameof(clientName)); _logger = logger ?? throw new ArgumentNullException(nameof(logger)); _messageHandlers = new ConcurrentDictionary(); _messageHandlersByName = new ConcurrentDictionary(); _generalMessageId = 0; _logGetMessageId = -1; // 初始化为-1,表示未开始 _logger.LogInformation("[{ClientName}] 创建消息ID管理器,初始LogGet ID: {LogGetId}", _clientName, _logGetMessageId); } #endregion #region 公共方法 /// /// 生成通用消息ID /// public long GenerateGeneralMessageId(JObject message, Action? callback = null, bool errorHandler = false) { ThrowIfDisposed(); ValidateMessage(message); var id = GetNextMessageId(); message["message_id"] = id; // 记录log_get消息的发送 var messageType = message["message"]?.ToString(); if (messageType == "log_get") { LogLogGetMessage(id, message); } // 注册回调处理器 if (callback != null) { _messageHandlers[id] = new MessageHandler { Callback = callback, ErrorHandler = errorHandler, CreatedAt = DateTime.UtcNow }; } _logger.LogDebug("[{ClientName}] 生成通用消息ID: {MessageId}", _clientName, id); return id; } /// /// 生成日志获取消息ID /// public long GenerateLogGetMessageId(JObject message, Action callback) { ThrowIfDisposed(); ValidateMessage(message); if (callback == null) throw new ArgumentNullException(nameof(callback)); // 生成新的消息ID var newLogGetId = GetNextMessageId(); message["message_id"] = newLogGetId; // 注册回调处理器 _messageHandlers[newLogGetId] = new MessageHandler { Callback = callback, ErrorHandler = false, CreatedAt = DateTime.UtcNow, IsLogGetHandler = true }; // 设置新的LogGet ID var oldLogGetId = Interlocked.Exchange(ref _logGetMessageId, newLogGetId); // 触发事件 LogGetIdChanged?.Invoke(this, new LogGetIdChangedEventArgs(oldLogGetId, newLogGetId)); _logger.LogDebug("[{ClientName}] LogGet ID变化: {OldId} -> {NewId}", _clientName, oldLogGetId, newLogGetId); return newLogGetId; } /// /// 处理消息响应 /// public bool HandleMessageResponse(JObject response, Action? errorHandler = null) { ThrowIfDisposed(); if (response == null) return false; // 检查消息处理器 var id = response["message_id"]?.Value(); if (id.HasValue && _messageHandlers.TryGetValue(id.Value, out var handler)) { return HandleMessageHandler(id.Value, handler, response, errorHandler); } // 检查按名称的消息处理器 var name = response["message"]?.ToString(); if (!string.IsNullOrEmpty(name) && _messageHandlersByName.TryGetValue(name, out var nameHandler)) { return HandleNamedMessageHandler(name, nameHandler, response); } return false; } /// /// 设置消息处理器 /// public void SetMessageHandler(string[] names, MessageHandler handler) { ThrowIfDisposed(); if (names == null || names.Length == 0) throw new ArgumentException("消息名称不能为空", nameof(names)); if (handler == null) throw new ArgumentNullException(nameof(handler)); foreach (var name in names) { if (!string.IsNullOrEmpty(name)) { _messageHandlersByName[name] = handler; } } _logger.LogDebug("[{ClientName}] 设置消息处理器: {Names}", _clientName, string.Join(", ", names)); } /// /// 取消消息处理器 /// public void UnsetMessageHandler(string[] names) { ThrowIfDisposed(); if (names == null || names.Length == 0) return; foreach (var name in names) { if (!string.IsNullOrEmpty(name)) { _messageHandlersByName.TryRemove(name, out _); } } _logger.LogDebug("[{ClientName}] 取消消息处理器: {Names}", _clientName, string.Join(", ", names)); } /// /// 检查是否为当前日志获取消息 /// public bool IsCurrentLogGetMessage(long messageId) { var currentLogGetId = Interlocked.Read(ref _logGetMessageId); return messageId == currentLogGetId; } /// /// 重置日志获取ID /// public void ResetLogGetId() { var oldLogGetId = Interlocked.Exchange(ref _logGetMessageId, -1); LogGetIdChanged?.Invoke(this, new LogGetIdChangedEventArgs(oldLogGetId, -1)); _logger.LogDebug("[{ClientName}] 重置LogGet ID: {OldId} -> -1", _clientName, oldLogGetId); } /// /// 清理过期的消息处理器 /// public int CleanupExpiredHandlers(int maxAge = 30000) // 默认30秒 { ThrowIfDisposed(); var now = DateTime.UtcNow; var expiredKeys = new List(); foreach (var kvp in _messageHandlers) { if (kvp.Value.CreatedAt.AddMilliseconds(maxAge) < now) { expiredKeys.Add(kvp.Key); } } var cleanedCount = 0; foreach (var key in expiredKeys) { if (_messageHandlers.TryRemove(key, out _)) { cleanedCount++; } } if (cleanedCount > 0) { _logger.LogDebug("[{ClientName}] 清理了 {CleanedCount} 个过期的消息处理器", _clientName, cleanedCount); // 触发清理事件 HandlerCleanup?.Invoke(this, new HandlerCleanupEventArgs( _messageHandlers.Count, _messageHandlersByName.Count, cleanedCount)); } return cleanedCount; } #endregion #region 私有方法 /// /// 获取下一个消息ID /// private long GetNextMessageId() { var id = Interlocked.Increment(ref _generalMessageId); // 检查溢出 if (id <= 0) { _logger.LogWarning("[{ClientName}] 消息ID溢出,重置为1", _clientName); Interlocked.Exchange(ref _generalMessageId, 1); return 1; } return id; } /// /// 验证消息对象 /// private void ValidateMessage(JObject message) { if (message == null) throw new ArgumentNullException(nameof(message)); } /// /// 处理消息处理器 /// private bool HandleMessageHandler(long id, MessageHandler handler, JObject response, Action? errorHandler) { // 如果不是通知消息,则移除处理器 if (response["notification"]?.Value() != true) { _messageHandlers.TryRemove(id, out _); } // 处理错误 if (response["error"] != null) { if (!handler.ErrorHandler) { errorHandler?.Invoke(response["error"]?.ToString() ?? "未知错误"); } else { handler.Callback?.Invoke(response); } return true; } // 正常处理 handler.Callback?.Invoke(response); return true; } /// /// 处理按名称的消息处理器 /// private bool HandleNamedMessageHandler(string name, MessageHandler handler, JObject response) { handler.Callback?.Invoke(response); return true; } /// /// 记录LogGet消息 /// private void LogLogGetMessage(long id, JObject message) { _logBuilder.Clear(); _logBuilder.AppendFormat("[{0}] 发送log_get消息: message_id={1}", _clientName, id); if (message["timeout"] != null) _logBuilder.AppendFormat(", timeout={0}", message["timeout"]); if (message["headers"] != null) _logBuilder.AppendFormat(", headers={0}", message["headers"]); _logBuilder.AppendFormat(", ThreadId={0}", Thread.CurrentThread.ManagedThreadId); _logger.LogDebug(_logBuilder.ToString()); } /// /// 检查是否已释放 /// private void ThrowIfDisposed() { if (_disposed) { throw new ObjectDisposedException(nameof(MessageIdManager)); } } #endregion #region IDisposable /// /// 释放资源 /// public void Dispose() { if (_disposed) return; _disposed = true; // 清理所有消息处理器 _messageHandlers.Clear(); _messageHandlersByName.Clear(); _logger.LogInformation("[{ClientName}] 释放消息ID管理器", _clientName); } #endregion } }