using System; using System.Collections.Generic; using System.Linq; using System.Runtime.CompilerServices; using System.Threading.Tasks; using DeviceDetectorNET.Parser.Device; using JiShe.CollectBus.Ammeters; using JiShe.CollectBus.Common.Consts; using JiShe.CollectBus.Common.Enums; using JiShe.CollectBus.Common.Extensions; using JiShe.CollectBus.Common.Helpers; using JiShe.CollectBus.Enums; using JiShe.CollectBus.Interceptors; using JiShe.CollectBus.IotSystems.Ammeters; using JiShe.CollectBus.IotSystems.Devices; using JiShe.CollectBus.IotSystems.MessageReceiveds; using JiShe.CollectBus.Kafka.Producer; using JiShe.CollectBus.Protocol.Contracts; using JiShe.CollectBus.Protocol.Contracts.Abstracts; using JiShe.CollectBus.Protocol.Contracts.Interfaces; using JiShe.CollectBus.Protocol.Contracts.Models; using JiShe.CollectBus.Protocol.Contracts.Services; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using TouchSocket.Core; using TouchSocket.Sockets; using Volo.Abp.Caching; using Volo.Abp.DependencyInjection; using Volo.Abp.Domain.Entities; using Volo.Abp.Domain.Repositories; using static System.Formats.Asn1.AsnWriter; using static FreeSql.Internal.GlobalFilter; namespace JiShe.CollectBus.Plugins { public partial class TcpMonitor : PluginBase, ITransientDependency, ITcpReceivedPlugin, ITcpConnectingPlugin, ITcpConnectedPlugin, ITcpClosedPlugin { private readonly IProducerService _producerService; private readonly ILogger _logger; private readonly IRepository _deviceRepository; private readonly IDistributedCache _ammeterInfoCache; private readonly IServiceProvider _serviceProvider; private readonly IProtocolService _protocolService; /// /// /// /// /// /// /// /// public TcpMonitor(IProducerService producerService, ILogger logger, IRepository deviceRepository, IDistributedCache ammeterInfoCache, IServiceProvider serviceProvider, IProtocolService protocolService) { _producerService = producerService; _logger = logger; _deviceRepository = deviceRepository; _ammeterInfoCache = ammeterInfoCache; _serviceProvider= serviceProvider; _protocolService = protocolService; } public async Task OnTcpReceived(ITcpSession client, ReceivedDataEventArgs e) { var messageHexString = Convert.ToHexString(e.ByteBlock.Span); var protocolPlugin = await _protocolService.GetProtocolServiceAsync("376.1"); if (protocolPlugin == null) { _logger.LogError("协议不存在!"); } var tcpSessionClient = (ITcpSessionClient)client; TB3761? tB3761 = await protocolPlugin!.AnalyzeAsync(tcpSessionClient, messageHexString); if (tB3761 == null) { // TODO: 暂时不处理,后续再处理 _logger.LogError($"指令初步解析失败,指令内容:{messageHexString}"); } await e.InvokeNext(); } //[GeneratorPlugin(typeof(ITcpConnectingPlugin))] public async Task OnTcpConnecting(ITcpSession client, ConnectingEventArgs e) { var tcpSessionClient = (ITcpSessionClient)client; _logger.LogInformation($"[TCP] ID:{tcpSessionClient.Id} IP:{client.GetIPPort()}正在连接中..."); await e.InvokeNext(); } //[GeneratorPlugin(typeof(ITcpConnectedPlugin))] public async Task OnTcpConnected(ITcpSession client, ConnectedEventArgs e) { var tcpSessionClient = (ITcpSessionClient)client; _logger.LogInformation($"[TCP] ID:{tcpSessionClient.Id} IP:{client.GetIPPort()}已连接"); await e.InvokeNext(); } //[GeneratorPlugin(typeof(ITcpClosedPlugin))]//ITcpSessionClient public async Task OnTcpClosed(ITcpSession client, ClosedEventArgs e) { var tcpSessionClient = (ITcpSessionClient)client; var entity = await _deviceRepository.FindAsync(a => a.ClientId == tcpSessionClient.Id); if (entity != null) { entity.UpdateByOnClosed(); await _deviceRepository.UpdateAsync(entity); } else { _logger.LogWarning($"[TCP] ID:{tcpSessionClient.Id} IP:{client.GetIPPort()}已关闭连接,但采集程序检索失败"); } await e.InvokeNext(); } } }