using Abp; using DotNetty.Buffers; using DotNetty.Transport.Channels; using MySqlX.XDevAPI; using System; using System.Collections; using System.Collections.Concurrent; using System.Collections.Generic; using System.ComponentModel.Design; using System.Diagnostics; using System.Text; using System.Threading.Tasks; using Yunda.ISAS.DataMonitoringServer.DataAnalysis; using Yunda.SOMS.DataMonitoringServer.FTPHandle; namespace Yunda.SOMS.DataMonitoringServer.TcpSocket.Server { using static ConstValue; public class DotNettyServerHandler : SimpleChannelInboundHandler { private IChannelHandlerContext _context; // 定义事件,事件处理程序包含消息和功能描述 private readonly ConcurrentDictionary _connections; private readonly Action _onMessageReceived; public DotNettyServerHandler(ConcurrentDictionary connections, Action onMessageReceived) { _connections = connections; _onMessageReceived = onMessageReceived; } public override void ChannelActive(IChannelHandlerContext context) { try { string clientIp = context.Channel.RemoteAddress.ToString(); MonitoringEventBus.LogHandler($"客户端连接:{clientIp}", "103客户端发送消息"); base.ChannelActive(context); } catch (Exception ex) { MonitoringEventBus.LogHandler($"{ex.StackTrace}", "103客户端错误信息"); } } public override void ChannelInactive(IChannelHandlerContext context) { try { string clientIp = context.Channel.RemoteAddress.ToString(); MonitoringEventBus.LogHandler($"客户端断开连接:{clientIp}", "103客户端发送消息"); base.ChannelInactive(context); } catch (Exception ex) { MonitoringEventBus.LogHandler($"{ex.StackTrace}", "103客户端错误信息"); } } protected override void ChannelRead0(IChannelHandlerContext ctx, IByteBuffer msg) { _context = ctx; try { byte startByte = msg.ReadByte(); ushort length = msg.ReadUnsignedShort(); byte address = msg.ReadByte(); byte controlWord = msg.ReadByte(); byte functionType = msg.ReadByte(); // 计算数据长度 int dataLength = length - 6; if (msg.ReadableBytes < dataLength) { // 数据不足,等待更多数据 return; } byte[] data = new byte[dataLength]; msg.ReadBytes(data); string clientIp = ctx.Channel.RemoteAddress.ToString(); string description = FunctionCodeDescriptions.GetDescription(functionType); //MonitoringEventBus.LogHandler($"装置地址 {address} 功能码:{functionType} 数据长度:{dataLength}", "103客户端消息"); Task.Run(() => _onMessageReceived?.Invoke(address, data, functionType)); HandleFunctionCodeAsync(functionType, address, controlWord, data, ctx); } catch (Exception ex) { MonitoringEventBus.LogHandler($"Error in ChannelRead0: {ex.StackTrace}", "103客户端发送消息"); } } private async Task HandleFunctionCodeAsync(byte functionType, byte address, byte controlWord, byte[] data, IChannelHandlerContext ctx) { switch (functionType) { case 0: HandleConfirmationMessage(address, data, ctx); break; case 1: MonitoringEventBus.LogHandler($"收到用户定值信息: {BitConverter.ToString(data)}", "103客户端信息"); break; case 2: // 厂家定值信息处理 break; case 3: MonitoringEventBus.LogHandler($"版本信息: {Encoding.ASCII.GetString(data)}", "103客户端信息"); break; case 4: // 自检信息处理 break; case 5: MonitoringEventBus.LogHandler($"开入开出信息: {Encoding.ASCII.GetString(data)}", "103客户端信息"); break; case 6: await UpdateDeviceCommunicationStateAsync(address, data, ctx); break; case 7: // B码对时状态处理 Debug.WriteLine($"B码对时状态处理: {BitConverter.ToString(data)}"); break; default: MonitoringEventBus.LogHandler("未知的功能码", "103客户端信息"); break; } } private void HandleConfirmationMessage(byte address, byte[] data, IChannelHandlerContext ctx) { Debug.WriteLine($"确认开入开出信息: {Encoding.ASCII.GetString(data)}"); // 可以添加发送确认报文的逻辑 } private async Task UpdateDeviceCommunicationStateAsync(byte address, byte[] data, IChannelHandlerContext ctx) { await Task.Run(async () => { try { if (DateTime.Now.Hour == 0&&DateTime.Now.Minute == 5&& DateTime.Now.Second<30) { await SendCustomMessageAsync(ctx, address, 0, 4, 0);//发送版本信息请求 } UpdateConnectionContext(address, ctx); await UpdateCommunicationStateCountAsync(address, ctx); } catch (Exception ex) { MonitoringEventBus.LogHandler($"Error: {ex.StackTrace}", "103客户端发送消息"); } }); } private void UpdateConnectionContext(byte address, IChannelHandlerContext ctx) { _connections[address] = ctx; } private async Task UpdateCommunicationStateCountAsync(byte address, IChannelHandlerContext ctx) { if (_connections.TryGetValue(address, out _)) { //_communicationStateCounts[address]++; //已经存在的装置地址 } else { //新加入的装置的地址 await SendInitMsgToDeviceAsync(ctx, address); } } private async Task SendInitMsgToDeviceAsync(IChannelHandlerContext ctx, byte address) { for (byte i = 1; i < 6; i++) { await SendCustomMessageAsync(ctx, address, 0, 5, i); } await SendCustomMessageAsync(ctx, address, 0, 1, 0); await SendCustomMessageAsync(ctx, address, 0, 2, 0); await SendCustomMessageAsync(ctx, address, 0, 3, 0); await SendCustomMessageAsync(ctx, address, 0, 4, 0); await SendCustomMessageAsync(ctx, address, 0, 7, 0); } public override async void ExceptionCaught(IChannelHandlerContext context, Exception ex) { try { MonitoringEventBus.LogHandler(ex.Message, "103客户端错误消息"); await context.CloseAsync(); } catch (Exception ex1) { MonitoringEventBus.LogHandler($"Error: {ex1.StackTrace}", "关闭tcp客户端"); } } // 发送自定义消息 public async Task SendCustomMessageAsync(IChannelHandlerContext context, byte address, byte controlWord, byte functionType, byte data) { try { // 构建数据帧 byte[] buffer = new byte[7]; // 填入启动字符 buffer[0] = 0x16; // 填入长度(假设数据部分为50字节,2字节长度字段+1字节地址+1字节标志位) byte dataLength = 7; buffer[1] = 0; // 长度低字节 buffer[2] = dataLength; // 长度高字节 // 填入地址、应用控制字、功能类型 buffer[3] = address; // 示例地址 buffer[4] = controlWord; // 示例应用控制字 buffer[5] = functionType; // 示例功能类型 buffer[6] = data; // 填入数据部分 IByteBuffer wrappedBuffer = Unpooled.WrappedBuffer(buffer); // 发送数据 await context.WriteAndFlushAsync(wrappedBuffer); string hexString = BitConverter.ToString(buffer).Replace("-", " "); MonitoringEventBus.LogHandler($"地址:{address} 功能码:{functionType} 数据:{hexString}", "103客户端发送消息"); await Task.Delay(1000); } catch (Exception ex) { MonitoringEventBus.LogHandler($"Error: {ex.StackTrace}", "103客户端发送消息"); } } } }