JiShe.CollectBus/JiShe.CollectBus.Core/Plugins/TcpMonitor.cs

using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.Common.Extensions;
using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.MongoDB;
using JiShe.CollectBus.MQ.Sender;
using MassTransit;
using Microsoft.Extensions.Logging;
using TouchSocket.Core;
using TouchSocket.Sockets;

namespace JiShe.CollectBus.Network.Plugins
{
    public partial class TcpMonitor(
        INSender nSender,
        ILogger<TcpMonitor> logger,
        IMongoRepository<MessageReceivedHeartbeatEvent> mongoHeartbeatRepository,
        IMongoRepository<MessageReceivedLoginEvent> mongoLoginRepository)
        : PluginBase
    {
        [GeneratorPlugin(typeof(ITcpReceivedPlugin))]
        public async Task OnTcpReceived(ITcpSessionClient client, ReceivedDataEventArgs e)
        {
            var messageHexString = Convert.ToHexString(e.ByteBlock.Span);
            var hexStringList = messageHexString.StringToPairs();
            var aFn = (int?)hexStringList.GetAnalyzeValue(CommandChunkEnum.AFN);
            var fn = (int?)hexStringList.GetAnalyzeValue(CommandChunkEnum.FN);
            var aTuple = (Tuple<string, int>)hexStringList.GetAnalyzeValue(CommandChunkEnum.A);

            if (aFn.HasValue && fn.HasValue)
            {
                if ((AFN)aFn == AFN.链路接口检测)
                {
                    switch (fn)
                    {
                        case 1://登录

                            var messageReceivedLoginEvent = new MessageReceivedLoginEvent
                            {
                                ClientId = client.Id,
                                ClientIp = client.IP,
                                ClientPort = client.Port,
                                MessageHexString = messageHexString,
                                DeviceNo = aTuple.Item1,
                                MessageId = NewId.NextGuid().ToString()
                            };
                            await mongoLoginRepository.AddAsync(messageReceivedLoginEvent);
                            await nSender.SendToReceivedLoginAsync(messageReceivedLoginEvent);
                            break;
                        case 3://心跳
                            var messageReceivedHeartbeatEvent = new MessageReceivedHeartbeatEvent
                            {
                                ClientId = client.Id,
                                ClientIp = client.IP,
                                ClientPort = client.Port,
                                MessageHexString = messageHexString,
                                DeviceNo = aTuple.Item1,
                                MessageId = NewId.NextGuid().ToString()
                            };
                            await mongoHeartbeatRepository.AddAsync(messageReceivedHeartbeatEvent);
                            await nSender.SendToReceivedHeartbeatAsync(messageReceivedHeartbeatEvent);
                            break;
                        default:
                            logger.LogError($"指令初步解析失败，指令内容：{messageHexString}");
                            break;
                    }
                }
                else
                {
                    await nSender.SendToReceivedAsync(new MessageReceivedEvent
                    {
                        ClientId = client.Id,
                        ClientIp = client.IP,
                        ClientPort = client.Port,
                        MessageHexString = messageHexString,
                        DeviceNo = aTuple.Item1,
                        MessageId = NewId.NextGuid().ToString()
                    });
                }
            }
            else
            {
                logger.LogError($"指令初步解析失败，指令内容：{messageHexString}");
            }

            await e.InvokeNext();
        }

        [GeneratorPlugin(typeof(ITcpConnectingPlugin))]
        public async Task OnTcpConnecting(ITcpSessionClient client,ConnectingEventArgs e)
        {
            logger.LogInformation($"[TCP] ID:{client.Id} IP:{client.GetIPPort()}正在连接中...");
            await e.InvokeNext();
        }

        [GeneratorPlugin(typeof(ITcpConnectedPlugin))]
        public async Task OnTcpConnected(ITcpSessionClient client,ConnectedEventArgs e)
        {
            logger.LogInformation($"[TCP] ID:{client.Id} IP:{client.GetIPPort()}已连接");
            await e.InvokeNext();
        }

        [GeneratorPlugin(typeof(ITcpClosedPlugin))]
        public async Task OnTcpClosed(ITcpSessionClient client, ClosedEventArgs e)
        {
            logger.LogInformation($"[TCP] ID:{client.Id} IP:{client.GetIPPort()}已关闭连接");
            await e.InvokeNext();
        }
    }
}