admin/JiShe.CollectBus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

提交

Browse Source
This commit is contained in:
zenghongyao 2025-04-23 15:10:20 +08:00
parent 9c24f53eb6
commit 8da5fee0a2
4 changed files with 50 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
protocols/JiShe.CollectBus.Protocol.Contracts/Interfaces/IProtocolService.cs
View File

@ -18,5 +18,13 @@ namespace JiShe.CollectBus.Protocol.Contracts.Interfaces
/// <returns></returns>
/// <exception cref="UserFriendlyException"></exception>
Task<ProtocolInfo> FirstOrDefaultByDeviceAsync(string deviceCode, bool isSpecial = false);
/// <summary>
/// 获取协议池服务
/// </summary>
/// <param name="deviceCode"></param>
/// <param name="isSpecial"></param>
/// <returns></returns>
Task<IProtocolPlugin?> GetProtocolServiceAsync(string deviceCode, bool isSpecial = false);
}
}

25
protocols/JiShe.CollectBus.Protocol.Contracts/Services/ProtocolService.cs
View File

@ -18,10 +18,12 @@ namespace JiShe.CollectBus.Protocol.Contracts.Services
public class ProtocolService : IProtocolService, ISingletonDependency
{
private readonly IFreeRedisProvider _freeRedisProvider;
private readonly IServiceProvider _serviceProvider;
public ProtocolService(IFreeRedisProvider freeRedisProvider)
public ProtocolService(IFreeRedisProvider freeRedisProvider, IServiceProvider serviceProvider)
{
_freeRedisProvider = freeRedisProvider;
_serviceProvider = serviceProvider;
}
/// <summary>
@ -42,6 +44,27 @@ namespace JiShe.CollectBus.Protocol.Contracts.Services
return protocolInfo;
}
/// <summary>
/// 获取协议池服务
/// </summary>
/// <param name="deviceCode"></param>
/// <param name="isSpecial"></param>
/// <returns></returns>
public async Task<IProtocolPlugin?> GetProtocolServiceAsync(string deviceCode, bool isSpecial = false)
{
try
{
ProtocolInfo protocolInfo= await FirstOrDefaultByDeviceAsync(deviceCode, isSpecial);
if(protocolInfo==null)
return null;
return _serviceProvider.GetKeyedService<IProtocolPlugin>(protocolInfo.Name);
}
catch (Exception ex)
{
return null;
}
}
private static bool ContainsExactPartRegex(string searchPattern, string fullString)
{
// 构建正则表达式 - 匹配以逗号或开头为边界，以逗号或结尾为边界的部分

18
services/JiShe.CollectBus.Application/Plugins/TcpMonitor.cs
View File

@ -18,6 +18,7 @@ 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;
@ -38,6 +39,7 @@ namespace JiShe.CollectBus.Plugins
private readonly IRepository<Device, Guid> _deviceRepository;
private readonly IDistributedCache<AmmeterInfo> _ammeterInfoCache;
private readonly IServiceProvider _serviceProvider;
private readonly IProtocolService _protocolService;
/// <summary>
///
@ -50,13 +52,14 @@ namespace JiShe.CollectBus.Plugins
public TcpMonitor(IProducerService producerService,
ILogger<TcpMonitor> logger,
IRepository<Device, Guid> deviceRepository,
IDistributedCache<AmmeterInfo> ammeterInfoCache, IServiceProvider serviceProvider)
IDistributedCache<AmmeterInfo> ammeterInfoCache, IServiceProvider serviceProvider, IProtocolService protocolService)
{
_producerService = producerService;
_logger = logger;
_deviceRepository = deviceRepository;
_ammeterInfoCache = ammeterInfoCache;
_serviceProvider= serviceProvider;
_protocolService = protocolService;
}
@ -64,14 +67,12 @@ namespace JiShe.CollectBus.Plugins
public async Task OnTcpReceived(ITcpSession client, ReceivedDataEventArgs e)
{
var messageHexString = Convert.ToHexString(e.ByteBlock.Span);
var tcpSessionClient = (ITcpSessionClient)client;
var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
var protocolPlugin = await _protocolService.GetProtocolServiceAsync("376.1");
if (protocolPlugin == null)
{
_logger.LogError("协议不存在！");
}
var tcpSessionClient = (ITcpSessionClient)client;
TB3761? tB3761 = await protocolPlugin!.AnalyzeAsync<TB3761>(tcpSessionClient, messageHexString);
if (tB3761 == null)
{
@ -131,13 +132,6 @@ namespace JiShe.CollectBus.Plugins
/// <returns></returns>
private async Task OnTcpNormalReceived(ITcpSessionClient tcpSessionClient,string messageHexString, TB3761? tB3761)
{
//var _protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
//if (_protocolPlugin == null)
//{
// _logger.LogError("376.1协议插件不存在！");
//}
//await _producerBus.Publish(new MessageReceived
//{
// ClientId = client.Id,

5
services/JiShe.CollectBus.Application/Subscribers/SubscriberAppService.cs
View File

@ -24,6 +24,7 @@ using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.IoTDB.Provider;
using JiShe.CollectBus.Protocol.Dto;
using System.Collections;
using JiShe.CollectBus.Interfaces;
namespace JiShe.CollectBus.Subscribers
{
@ -36,6 +37,7 @@ namespace JiShe.CollectBus.Subscribers
private readonly IRepository<MessageReceivedHeartbeat, Guid> _messageReceivedHeartbeatEventRepository;
private readonly IMeterReadingRecordRepository _meterReadingRecordsRepository;
private readonly IIoTDbProvider _dbProvider;
private readonly IProtocolService _protocolService;
/// <summary>
/// Initializes a new instance of the <see cref="SubscriberAppService"/> class.
@ -52,7 +54,7 @@ namespace JiShe.CollectBus.Subscribers
IRepository<MessageReceivedLogin, Guid> messageReceivedLoginEventRepository,
IRepository<MessageReceivedHeartbeat, Guid> messageReceivedHeartbeatEventRepository,
IIoTDbProvider dbProvider,
IMeterReadingRecordRepository meterReadingRecordsRepository)
IMeterReadingRecordRepository meterReadingRecordsRepository, IProtocolService protocolService)
{
_logger = logger;
_tcpService = tcpService;
@ -61,6 +63,7 @@ namespace JiShe.CollectBus.Subscribers
_messageReceivedHeartbeatEventRepository = messageReceivedHeartbeatEventRepository;
_meterReadingRecordsRepository = meterReadingRecordsRepository;
_dbProvider = dbProvider;
_protocolService = protocolService;
}
[LogIntercept]