TestLogs/LTEMvcApp/Services/LTEClientWebSocket.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using WebSocket4Net;
using LTEMvcApp.Models;
using System.Text.RegularExpressions;
using Microsoft.Extensions.Logging;

namespace LTEMvcApp.Services
{
    /// <summary>
    /// LTE客户端WebSocket实现 - 对应JavaScript中的lte.client.server
    /// </summary>
    public class LTEClientWebSocket : IDisposable
    {
        #region 私有字段

        private WebSocket? _webSocket;
        private readonly LTEClient _client;
        private readonly ClientConfig _config;
        private readonly ConcurrentDictionary<int, MessageHandler> _messageHandlers;
        private readonly ConcurrentDictionary<string, MessageHandler> _messageHandlersByName;
        private readonly ConcurrentQueue<JObject> _messageFifo;
        private readonly CancellationTokenSource _cancellationTokenSource;
        private Timer? _reconnectTimer;
        private Timer? _statsTimer;
        private Timer? _messageDeferTimer;
        private Timer? _readyTimer;
        private int _messageId;
        private int _logGetId;
        private bool _disposed;
        private LogParserService logParser = new LogParserService();
        private readonly ILogger<LTEClientWebSocket> _logger;
        #endregion

        #region 事件

        /// <summary>
        /// 连接打开事件
        /// </summary>
        public event EventHandler? ConnectionOpened;

        /// <summary>
        /// 连接关闭事件
        /// </summary>
        public event EventHandler? ConnectionClosed;

        /// <summary>
        /// 连接错误事件
        /// </summary>
        public event EventHandler<string>? ConnectionError;

        /// <summary>
        /// 消息接收事件
        /// </summary>
        public event EventHandler<JObject>? MessageReceived;

        /// <summary>
        /// 日志接收事件
        /// </summary>
        public event EventHandler<List<LTELog>>? LogsReceived;

        /// <summary>
        /// 统计信息接收事件
        /// </summary>
        public event EventHandler<JObject>? StatsReceived;

        /// <summary>
        /// 状态改变事件
        /// </summary>
        public event EventHandler<ClientState>? StateChanged;

        /// <summary>
        /// 客户端网格刷新事件
        /// </summary>
        public event EventHandler? ClientGridRefreshed;

        #endregion

        #region 属性

        /// <summary>
        /// 客户端
        /// </summary>
        public LTEClient Client => _client;

        /// <summary>
        /// 配置
        /// </summary>
        public ClientConfig Config => _config;

        /// <summary>
        /// 是否已连接
        /// </summary>
        public bool IsConnected => _webSocket?.State == WebSocketState.Open;

        /// <summary>
        /// 当前状态
        /// </summary>
        public ClientState State => _client.State;

        /// <summary>
        /// 是否暂停
        /// </summary>
        public bool IsPaused => _config.Pause;

        /// <summary>
        /// 是否只读
        /// </summary>
        public bool IsReadonly => _config.Readonly;

        #endregion

        #region 构造函数

        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="config">客户端配置</param>
        /// <param name="logger">ILogger实例</param>
        public LTEClientWebSocket(ClientConfig config, ILogger<LTEClientWebSocket> logger)
        {
            _config = config;
            _client = new LTEClient(config);
            _messageHandlers = new ConcurrentDictionary<int, MessageHandler>();
            _messageHandlersByName = new ConcurrentDictionary<string, MessageHandler>();
            _messageFifo = new ConcurrentQueue<JObject>();
            _cancellationTokenSource = new CancellationTokenSource();
            _messageId = 0;
            _logger = logger;
            _logger.LogInformation($"创建WebSocket客户端: {config.Name}");
        }

        #endregion

        #region 公共方法

        /// <summary>
        /// 启动WebSocket连接
        /// </summary>
        public void Start()
        {
            if (_disposed) return;

            try
            {
                _logger.LogInformation($"[{_config.Name}] 尝试连接: {_config.Address}");
                SetState(ClientState.Connecting);

                // 构建WebSocket URL
                var address = _config.Address ?? "192.168.13.12:9001";
                var url = (_config.Ssl ? "wss://" : "ws://") + address;

                _webSocket = new WebSocket(url);
                _webSocket.EnableAutoSendPing = false;

                // 绑定事件处理器
                _webSocket.Opened += OnSocketOpened;
                _webSocket.Closed += OnSocketClosed;
                _webSocket.MessageReceived += OnSocketMessage0;
                _webSocket.Error += OnSocketError;

                _webSocket.Open();
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, $"[{_config.Name}] 连接异常: {ex.Message}");
                ConnectionError?.Invoke(this, $"无法连接到 {_config.Address}: {ex.Message}");
                SetState(ClientState.Error);
            }
        }

        /// <summary>
        /// 停止WebSocket连接
        /// </summary>
        public void Stop()
        {
            SetState(ClientState.Stop);
            StopTimers();
        }

        /// <summary>
        /// 重置日志
        /// </summary>
        public void ResetLogs()
        {
            if (State == ClientState.Connected)
            {
                SendMessage(new JObject { ["message"] = "log_reset" }, response =>
                {
                    _client.ResetLogs();
                });
            }
            else
            {
                _client.ResetLogs();
            }
        }

        /// <summary>
        /// 切换播放/暂停状态
        /// </summary>
        public void PlayPause()
        {
            if (_config.Pause)
            {
                _config.Pause = false;
                if (State == ClientState.Connected)
                {
                    LogGet();
                }
            }
            else
            {
                _config.Pause = true;
            }
        }

        /// <summary>
        /// 设置日志配置
        /// </summary>
        /// <param name="config">日志配置</param>
        /// <param name="save">是否保存</param>
        public void SetLogsConfig(Dictionary<string, object> config, bool save = false)
        {
            // 重新格式化配置
            var logs = new JObject();
            foreach (var kvp in config)
            {
                switch (kvp.Key)
                {
                    case "layers":
                        var layers = new JObject();
                        var layersDict = kvp.Value as Dictionary<string, object>;
                        if (layersDict != null)
                        {
                            foreach (var layer in layersDict)
                            {
                                var layerConfig = layer.Value as Dictionary<string, object>;
                                if (layerConfig != null)
                                {
                                    layers[layer.Key] = new JObject
                                    {
                                        ["level"] = layerConfig.GetValueOrDefault("level", "warn").ToString(),
                                        ["max_size"] = Convert.ToInt32(layerConfig.GetValueOrDefault("max_size", 0)),
                                        ["payload"] = Convert.ToBoolean(layerConfig.GetValueOrDefault("payload", false))
                                    };
                                }
                            }
                        }
                        logs["layers"] = layers;
                        break;
                    default:
                        if (IsLogParameter(kvp.Key))
                        {
                            logs[kvp.Key] = JToken.FromObject(kvp.Value);
                        }
                        break;
                }
            }

            SendMessage(new JObject
            {
                ["message"] = "config_set",
                ["logs"] = logs
            }, response =>
            {
                if (save)
                {
                    // 保存配置
                    foreach (var kvp in config)
                    {
                        if (_config.Logs.ContainsKey(kvp.Key))
                        {
                            _config.Logs[kvp.Key] = kvp.Value;
                        }
                        else
                        {
                            _config.Logs.Add(kvp.Key, kvp.Value);
                        }
                    }
                }
                LogGet(new Dictionary<string, object> { ["timeout"] = 0 });
            }, true);
        }

        /// <summary>
        /// 设置消息处理器
        /// </summary>
        /// <param name="names">消息名称</param>
        /// <param name="handler">处理器</param>
        public void SetMessageHandler(string[] names, MessageHandler handler)
        {
            SendMessage(new JObject
            {
                ["message"] = "register",
                ["register"] = string.Join(",", names)
            });

            foreach (var name in names)
            {
                _messageHandlersByName[name] = handler;
            }
        }

        /// <summary>
        /// 取消消息处理器
        /// </summary>
        /// <param name="names">消息名称</param>
        public void UnsetMessageHandler(string[] names)
        {
            SendMessage(new JObject
            {
                ["message"] = "register",
                ["unregister"] = string.Join(",", names)
            });

            foreach (var name in names)
            {
                _messageHandlersByName.TryRemove(name, out _);
            }
        }

        /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="message">消息</param>
        /// <param name="callback">回调</param>
        /// <param name="errorHandler">错误处理器</param>
        /// <returns>消息ID</returns>
        public int SendMessage(JObject message, Action<JObject>? callback = null, bool errorHandler = false)
        {
            if (_webSocket?.State != WebSocketState.Open) return -1;
            if (message == null) return -1;

            var id = ++_messageId;
            message["message_id"] = id;

            if (callback != null)
            {
                _messageHandlers[id] = new MessageHandler
                {
                    Callback = callback,
                    ErrorHandler = errorHandler
                };
            }

            if (_messageDeferTimer != null)
            {
                _messageDeferTimer.Dispose();
                _messageDeferTimer = null;
            }

            _messageFifo.Enqueue(message);

            if (_messageFifo.Count < 100) // 批处理大小
            {
                _messageDeferTimer = new Timer(_ => SendMessageNow(), null, 1, Timeout.Infinite);
            }
            else
            {
                SendMessageNow();
            }

            return id;
        }

        /// <summary>
        /// 获取日志
        /// </summary>
        /// <param name="parameters">参数</param>
        public void LogGet(Dictionary<string, object>? parameters = null)
        {
            var config = _config;
            if (config.Pause)
                return;

            var layers = new JObject();
            if (config.Logs.ContainsKey("layers"))
            {
                // 处理layers配置，确保键是字符串类型
                if (config.Logs["layers"] is JObject layersJObject)
                {
                    foreach (var layer in layersJObject)
                    {
                        var layerName = layer.Key; // 这里layer.Key已经是字符串
                        var layerConfig = layer.Value as JObject;
                        if (layerConfig != null && layerConfig.ContainsKey("filter"))
                        {
                            layers[layerName] = layerConfig["filter"].ToString();
                        }
                    }
                }
                else if (config.Logs["layers"] is Dictionary<string, object> layersDict)
                {
                    foreach (var layer in layersDict)
                    {
                        var layerConfig = layer.Value as Dictionary<string, object>;
                        if (layerConfig != null && layerConfig.ContainsKey("filter"))
                        {
                            layers[layer.Key] = layerConfig["filter"].ToString();
                        }
                    }
                }
            }

            var message = new JObject
            {
                ["timeout"] = 1,
                ["min"] = 64,
                ["max"] = 2048,
                ["layers"] = layers,
                ["message"] = "log_get",
                ["headers"] = _client.LogCount == 0
            };

            if (parameters != null)
            {
                foreach (var param in parameters)
                {
                    message[param.Key] = JToken.FromObject(param.Value);
                }
            }

            _logGetId = SendMessage(message, LogGetParse);
        }

        #endregion

        #region 私有方法

        /// <summary>
        /// WebSocket连接打开事件
        /// </summary>
        private void OnSocketOpened(object? sender, EventArgs e)
        {
            _logger.LogInformation($"[{_config.Name}] WebSocket连接已打开");
            StopTimers();
            _readyTimer = new Timer(_ => OnSocketReady(), null, 2500, Timeout.Infinite);
            ConnectionOpened?.Invoke(this, EventArgs.Empty);
        }

        /// <summary>
        /// WebSocket连接关闭事件
        /// </summary>
        private void OnSocketClosed(object? sender, EventArgs e)
        {
            _logger.LogWarning($"[{_config.Name}] WebSocket连接已关闭");
            ConnectionClosed?.Invoke(this, EventArgs.Empty);

            StopTimers();
            CloseComponents();

            if (State == ClientState.Connected)
            {
                // 处理监控窗口停止
            }

            if (_config.Enabled)
            {
                Console.WriteLine("启动重连定时器");
                if (State != ClientState.Stop)
                {
                    SetState(ClientState.Error);
                }
                _reconnectTimer = new Timer(_ => Start(), null, _config.ReconnectDelay, Timeout.Infinite);
            }
        }

        /// <summary>
        /// WebSocket错误事件
        /// </summary>
        private void OnSocketError(object? sender, SuperSocket.ClientEngine.ErrorEventArgs e)
        {
            _logger.LogError(e.Exception, $"[{_config.Name}] WebSocket错误: {e.Exception.Message}");
            SetState(ClientState.Error);
            ConnectionError?.Invoke(this, e.Exception.Message);
        }

        /// <summary>
        /// 初始消息处理（对应_onSocketMessage0）
        /// </summary>
        private void OnSocketMessage0(object? sender, MessageReceivedEventArgs e)
        {
            _logger.LogDebug($"[{_config.Name}] 收到初始消息: {e.Message}");
            StopTimers();

            try
            {
                var data = e.Message;
                var msg = JObject.Parse(data);

                switch (msg["message"]?.ToString())
                {
                    case "authenticate":
                        if (msg["ready"]?.Value<bool>() == true)
                        {
                            OnSocketReady();
                        }
                        else if (msg["error"] == null && !string.IsNullOrEmpty(_config.Password))
                        {
                            // 重新认证
                            Authenticate(_config.Password, msg["challenge"]?.ToString() ?? "");
                        }
                        else
                        {
                            // 提示输入密码
                            PromptPassword(msg);
                        }
                        break;

                    case "ready":
                        OnSocketReady();
                        break;

                    default:
                        break;
                }
            }
            catch (Exception ex)
            {
                ConnectionError?.Invoke(this, $"JSON解析错误: {ex.Message}");
            }
        }

        /// <summary>
        /// WebSocket准备就绪（对应_onSocketReady）
        /// </summary>
        private void OnSocketReady()
        {
            if (_webSocket == null) return;

            // 切换到正常的消息处理函数
            _webSocket.MessageReceived -= OnSocketMessage0;
            _webSocket.MessageReceived += OnSocketMessage;

            _messageFifo.Clear();

            // 检查当前配置
            var firstCon = !_config.Logs.ContainsKey("layers");

            // 获取配置
            SendMessage(new JObject { ["message"] = "config_get" }, config =>
            {
                Console.WriteLine("配置已接收");

                _client.ResetLogs();

                // 设置基本信息
                _client.Version = config["version"]?.ToString();
                _client.Name = config["name"]?.ToString() ?? _config.Name;
                _client.Model = config["type"]?.ToString();

                if (config["profiling"]?.Value<bool>() == true)
                {
                    // 设置性能分析可用标志
                }

                var ro = _config.Readonly;
                if (ro || firstCon)
                {
                    _config.Logs = config["logs"]?.ToObject<Dictionary<string, object>>() ?? new Dictionary<string, object>();
                }
                else
                {
                    var serverLogs = config["logs"]?.ToObject<Dictionary<string, object>>() ?? new Dictionary<string, object>();
                    foreach (var kvp in serverLogs)
                    {
                        if (!_config.Logs.ContainsKey(kvp.Key))
                        {
                            _config.Logs[kvp.Key] = kvp.Value;
                        }
                    }
                }

                // 清理和设置层配置
                if (_config.Logs.ContainsKey("layers"))
                {
                    var layers = _config.Logs["layers"] as Dictionary<string, object>;
                    var configLayers = config["logs"]?["layers"]?.ToObject<Dictionary<string, object>>();

                    if (layers != null && configLayers != null)
                    {
                        var keysToRemove = layers.Keys.Where(k => !configLayers.ContainsKey(k)).ToList();
                        foreach (var key in keysToRemove)
                        {
                            layers.Remove(key);
                        }

                        foreach (var layer in layers)
                        {
                            var layerConfig = layer.Value as Dictionary<string, object>;
                            if (layerConfig != null && !layerConfig.ContainsKey("filter"))
                            {
                                if (ro)
                                {
                                    layerConfig["filter"] = layerConfig.GetValueOrDefault("level", "warn").ToString();
                                }
                                else
                                {
                                    layerConfig["filter"] = GetDefaultFilter(layer.Key);
                                }
                            }
                        }
                    }
                }

                // 添加小区信息
                if (config["cells"] != null)
                {
                    var cells = config["cells"]?.ToObject<Dictionary<string, object>>();
                    if (cells != null)
                    {
                        foreach (var cell in cells)
                        {
                            // 添加小区配置
                        }
                    }
                }

                SetState(ClientState.Connected);

                if (firstCon && !_config.SkipLogMenu)
                {
                    // 显示配置窗口
                    Console.WriteLine("请配置日志");
                }
                else if (ro)
                {
                    LogGet(new Dictionary<string, object> { ["timeout"] = 0 });
                }
                else
                {
                    SetLogsConfig(_config.Logs);
                }
            });
        }

        /// <summary>
        /// 正常消息处理（对应_onSocketMessage）
        /// </summary>
        private void OnSocketMessage(object? sender, MessageReceivedEventArgs e)
        {
            _logger.LogDebug($"[{_config.Name}] 收到消息: {e.Message}");
            try
            {
                var data = e.Message;
                JObject msg;

                if (data.StartsWith("["))
                {
                    // 处理二进制数据
                    var array = JArray.Parse(data);
                    msg = array[0] as JObject ?? new JObject();
                    // 处理二进制数据部分
                }
                else
                {
                    msg = JObject.Parse(data);
                }

                MessageReceived?.Invoke(this, msg);

                // 检查消息处理器
                var id = msg["message_id"]?.Value<int>();
                if (id.HasValue && _messageHandlers.TryGetValue(id.Value, out var handler))
                {
                    if (msg["notification"]?.Value<bool>() != true)
                    {
                        _messageHandlers.TryRemove(id.Value, out _);
                    }

                    if (msg["error"] != null)
                    {
                        if (!handler.ErrorHandler)
                        {
                            ConnectionError?.Invoke(this, msg["error"]?.ToString() ?? "未知错误");
                        }
                        else
                        {
                            handler.Callback?.Invoke(msg);
                        }
                        return;
                    }

                    handler.Callback?.Invoke(msg);
                    return;
                }

                // 检查按名称的消息处理器
                var name = msg["message"]?.ToString();
                if (!string.IsNullOrEmpty(name) && _messageHandlersByName.TryGetValue(name, out var nameHandler))
                {
                    nameHandler.Callback?.Invoke(msg);
                    return;
                }

                // 处理特定消息类型
                switch (name)
                {
                    case "log_get":
                        LogGetParse(msg);
                        break;
                    case "stats":
                        StatsReceived?.Invoke(this, msg);
                        break;
                    default:
                        Console.WriteLine($"未知消息: {name}");
                        break;
                }
            }
            catch (Exception ex)
            {
                ConnectionError?.Invoke(this, $"消息处理错误: {ex.Message}");
            }
        }

        /// <summary>
        /// 日志获取解析
        /// </summary>
        private void LogGetParse(JObject msg)
        {
            _logger.LogInformation($"[{_config.Name}] 解析日志消息: {msg}");
            var count = _client.LogCount;

            if (msg["headers"] != null)
            {
                var headers = msg["headers"]?.ToObject<string[]>();
                if (headers != null)
                {
                    _client.SetHeaders(headers);
                }
            }

            // 初始化模型猜测
            _client.LogModelGuessInit();

            var logs = msg["logs"];
            if (logs != null)
            {
                Console.WriteLine($"接收到日志: {logs.Count()} 条");
                if (logs is JArray logsArray)
                {
                    // 模型猜测
                    _client.LogModelGuess(logsArray.ToObject<List<Dictionary<string, object>>>() ?? new List<Dictionary<string, object>>());

                    var logList = new List<LTELog>();
                    foreach (var logItem in logsArray)
                    {
                        if (logItem is JObject logObj)
                        {
                            //var logData = logObj.ToObject<Dictionary<string, object>>() ?? new Dictionary<string, object>();

                            // 创建LTELog对象
                            var log = JsonConvert.DeserializeObject<LTELog>(logObj.ToString()); //new LTELog(logData);

                            // 处理消息和数据
                            if (log.Data is List<string> dataList && dataList.Count > 0)
                            {
                                log.Message = dataList[0];
                                dataList.RemoveAt(0);
                            }

                            // 设置方向
                            log.Direction = _client.DirConvert(log);

                            // 处理信息字段
                            if (log.Info != null)
                            {
                                log.Info = _client.StringToId(log.Info.ToString());
                            }

                            // 处理PHY层的信号记录
                            if (log.Layer == "PHY" && log.Data is List<string> data)
                            {
                                var signalRecord = new Dictionary<string, object>();
                                for (int j = data.Count - 1; j >= 0; j--)
                                {
                                    var line = data[j];
                                    var match = Regex.Match(line, @"Link:\s([\w\d]+)@(\d+)");
                                    if (match.Success)
                                    {
                                        var linkName = match.Groups[1].Value;
                                        var offset = uint.Parse(match.Groups[2].Value);
                                        signalRecord[linkName] = new { offset = offset };
                                        data.RemoveAt(j);
                                    }
                                }

                                if (signalRecord.Count > 0)
                                {
                                    //log.SignalRecord = signalRecord;
                                    _client.HasSignalRecord = true;
                                }
                            }

                            log.Client = _client;
                            logList.Add(log);
                        }
                    }
                    logParser.ParseLogList(_client, logList, true);
                    //_client.ParseLogList(logList, true);
                    LogsReceived?.Invoke(this, logList);
                }
            }

            if (count == 0 && _client.LogCount > 0)
            {
                // 刷新客户端网格
                ClientGridRefreshed?.Invoke(this, EventArgs.Empty);
            }

            // 更新日志获取 - 只有在最后一个请求时才更新
            if (msg["message_id"]?.Value<int>() == _logGetId)
            {
                LogGet();
            }
        }

        /// <summary>
        /// 发送消息
        /// </summary>
        private void SendMessageNow()
        {
            if (_webSocket?.State != WebSocketState.Open) return;

            var messages = new List<JObject>();
            while (_messageFifo.TryDequeue(out var message))
            {
                messages.Add(message);
            }

            if (messages.Count == 1)
            {
                var json = JsonConvert.SerializeObject(messages[0]);
                _webSocket.Send(json);
            }
            else
            {
                var json = JsonConvert.SerializeObject(messages);
                _webSocket.Send(json);
            }

            _messageDeferTimer?.Dispose();
            _messageDeferTimer = null;
        }

        /// <summary>
        /// 停止定时器
        /// </summary>
        private void StopTimers()
        {
            _reconnectTimer?.Dispose();
            _reconnectTimer = null;

            _statsTimer?.Dispose();
            _statsTimer = null;

            _messageDeferTimer?.Dispose();
            _messageDeferTimer = null;

            _readyTimer?.Dispose();
            _readyTimer = null;
        }

        /// <summary>
        /// 关闭组件
        /// </summary>
        private void CloseComponents()
        {
            // 关闭所有组件
        }

        /// <summary>
        /// 设置状态
        /// </summary>
        private void SetState(ClientState state)
        {
            if (_client.State != state)
            {
                _client.State = state;
                StateChanged?.Invoke(this, state);
            }
        }

        /// <summary>
        /// 获取默认过滤器
        /// </summary>
        private string GetDefaultFilter(string layer)
        {
            return layer switch
            {
                "NAS" or "RRC" => "debug",
                "EVENT" or "ALARM" or "MON" or "PROD" => "info",
                _ => "warn"
            };
        }

        /// <summary>
        /// 检查是否为日志参数
        /// </summary>
        private bool IsLogParameter(string parameter)
        {
            var logParams = new[] { "signal", "cch", "bcch", "mib", "rep", "csi", "dci_size", "cell_meas" };
            return logParams.Contains(parameter);
        }

        /// <summary>
        /// 认证
        /// </summary>
        private void Authenticate(string password, string challenge)
        {
            // 实现认证逻辑
            var authMessage = new JObject
            {
                ["message"] = "authenticate",
                ["password"] = password
            };
            SendMessage(authMessage);
        }

        /// <summary>
        /// 提示输入密码
        /// </summary>
        private void PromptPassword(JObject msg)
        {
            // 实现密码提示逻辑
            Console.WriteLine("需要认证，请输入密码");
        }

        #endregion

        #region IDisposable

        public void Dispose()
        {
            if (_disposed) return;

            _disposed = true;
            Stop();
            StopTimers();
            _cancellationTokenSource.Cancel();
            _cancellationTokenSource.Dispose();
            _webSocket?.Dispose();
            _logger.LogInformation($"[{_config.Name}] 释放WebSocket客户端");
        }

        #endregion
    }

    /// <summary>
    /// 消息处理器
    /// </summary>
    public class MessageHandler
    {
        public Action<JObject>? Callback { get; set; }
        public bool ErrorHandler { get; set; }
    }
}