81 lines
3.4 KiB
C#
81 lines
3.4 KiB
C#
using JiShe.CollectBus.Common.Consts;
|
|
using JiShe.CollectBus.Common.Helpers;
|
|
using JiShe.CollectBus.IoTDB.Interface;
|
|
using JiShe.CollectBus.IotSystems.MessageReceiveds;
|
|
using JiShe.CollectBus.Kafka.Attributes;
|
|
using JiShe.CollectBus.Kafka.Internal;
|
|
using JiShe.CollectBus.Protocol.Contracts;
|
|
using JiShe.CollectBus.Protocol.Interfaces;
|
|
using JiShe.CollectBus.Protocol3761;
|
|
using JiShe.CollectBus.Repository.MeterReadingRecord;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Logging;
|
|
using System;
|
|
using System.Threading.Tasks;
|
|
using JiShe.CollectBus.Protocol;
|
|
using TouchSocket.Sockets;
|
|
|
|
namespace JiShe.CollectBus.Subscribers
|
|
{
|
|
public class SubscriberAnalysisAppService : CollectBusAppService, ISubscriberAnalysisAppService, IKafkaSubscribe
|
|
{
|
|
private readonly ILogger<SubscriberAnalysisAppService> _logger;
|
|
private readonly ITcpService _tcpService;
|
|
private readonly IServiceProvider _serviceProvider;
|
|
private readonly IMeterReadingRecordRepository _meterReadingRecordsRepository;
|
|
private readonly IIoTDbProvider _dbProvider;
|
|
private readonly IProtocolService _protocolService;
|
|
|
|
public SubscriberAnalysisAppService(ILogger<SubscriberAnalysisAppService> logger,
|
|
ITcpService tcpService,
|
|
IServiceProvider serviceProvider,
|
|
IIoTDbProvider dbProvider,
|
|
IMeterReadingRecordRepository meterReadingRecordsRepository, IProtocolService protocolService)
|
|
{
|
|
_logger = logger;
|
|
_tcpService = tcpService;
|
|
_serviceProvider = serviceProvider;
|
|
_meterReadingRecordsRepository = meterReadingRecordsRepository;
|
|
_dbProvider = dbProvider;
|
|
_protocolService = protocolService;
|
|
}
|
|
|
|
/// <summary>
|
|
/// 解析AFN00H
|
|
/// </summary>
|
|
/// <param name="receivedMessage"></param>
|
|
/// <returns></returns>
|
|
[KafkaSubscribe(ProtocolConst.SubscriberAFN00HReceivedEventNameTemp)]
|
|
public async Task<ISubscribeAck> ReceivedAFN00Event(MessageProtocolAnalysis<TB3761> receivedMessage)
|
|
{
|
|
var protocolPlugin = await _protocolService.GetProtocolServiceAsync(receivedMessage.DeviceNo);
|
|
if (protocolPlugin == null)
|
|
{
|
|
_logger.LogError("协议不存在！");
|
|
}
|
|
else
|
|
{
|
|
if (receivedMessage.Data==null)
|
|
{
|
|
Logger.LogError($"数据解析失败：{receivedMessage.Serialize()}");
|
|
return SubscribeAck.Success();
|
|
}
|
|
if (receivedMessage.Data.DT == null || receivedMessage.Data.AFN_FC == null)
|
|
{
|
|
Logger.LogError($"数据解析失败：{receivedMessage.Serialize()}");
|
|
return SubscribeAck.Success();
|
|
}
|
|
string serverName = $"AFN{receivedMessage.Data.AFN_FC.AFN}_F{receivedMessage.Data.DT.Fn}_Analysis";
|
|
//var analysisStrategy = _serviceProvider.GetKeyedService<IAnalysisStrategy>($"AFN0_F1_Analysis");
|
|
|
|
//var data = await analysisStrategy.ExecuteAsync<UnitDataDto<bool>>(tB3761);
|
|
var executor = _serviceProvider.GetRequiredService<AnalysisStrategyContext>();
|
|
await executor.ExecuteAsync<TB3761, object>(serverName, receivedMessage.Data);
|
|
}
|
|
|
|
return SubscribeAck.Fail();
|
|
}
|
|
|
|
}
|
|
}
|