JiShe.CollectBus/services/JiShe.CollectBus.Application/Subscribers/SubscriberAnalysisAppService.cs

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;
using JiShe.CollectBus.Protocol.Dto;
using JiShe.CollectBus.Protocol.Contracts.Protocol.Dto;

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>(serverName, receivedMessage.Data);
            }

            return SubscribeAck.Fail();
        }

        /// <summary>
        /// 解析AFN01H
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.SubscriberAFN01HReceivedEventNameTemp)]
        public async Task<ISubscribeAck> ReceivedAFN01Event(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>(serverName, receivedMessage.Data);
            }

            return SubscribeAck.Fail();
        }


        /// <summary>
        /// 解析AFN02H
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.SubscriberAFN02HReceivedEventNameTemp)]
        public async Task<ISubscribeAck> ReceivedAFN02Event(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>(serverName, receivedMessage.Data, (result) =>
                {
                    var ssss = (UnitDataAnalysis<AnalysisBaseDto<string>>)result;
                    _logger.LogInformation($"解析AFN02H数据：{ssss.Serialize()}");
                });
                return SubscribeAck.Success();
            }
            return SubscribeAck.Fail();

        }


        /// <summary>
        /// 解析AFN03H
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.SubscriberAFN03HReceivedEventNameTemp)]
        public async Task<ISubscribeAck> ReceivedAFN03Event(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>(serverName, receivedMessage.Data);
                return SubscribeAck.Success();
            }

            return SubscribeAck.Fail();
        }


        /// <summary>
        /// 解析AFN04H
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.SubscriberAFN04HReceivedEventNameTemp)]
        public async Task<ISubscribeAck> ReceivedAFN04Event(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>(serverName, receivedMessage.Data);
                return SubscribeAck.Success();
            }

            return SubscribeAck.Fail();
        }


        /// <summary>
        /// 解析AFN05H
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.SubscriberAFN05HReceivedEventNameTemp)]
        public async Task<ISubscribeAck> ReceivedAFN05Event(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>(serverName, receivedMessage.Data);
                return SubscribeAck.Success();
            }

            return SubscribeAck.Fail();
        }


        /// <summary>
        /// 解析AFN09H
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.SubscriberAFN09HReceivedEventNameTemp)]
        public async Task<ISubscribeAck> ReceivedAFN09Event(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>(serverName, receivedMessage.Data);
                return SubscribeAck.Success();
            }

            return SubscribeAck.Fail();
        }

        /// <summary>
        /// 解析AFN0AH
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.SubscriberAFN0AHReceivedEventNameTemp)]
        public async Task<ISubscribeAck> ReceivedAFN0AEvent(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>(serverName, receivedMessage.Data);
                return SubscribeAck.Success();
            }

            return SubscribeAck.Fail();
        }

        /// <summary>
        /// 解析AFN0BH
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.SubscriberAFN0BHReceivedEventNameTemp)]
        public async Task<ISubscribeAck> ReceivedAFN0BEvent(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>(serverName, receivedMessage.Data);
                return SubscribeAck.Success();
            }

            return SubscribeAck.Fail();
        }

        /// <summary>
        /// 解析AFN0CH
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.SubscriberAFN0CHReceivedEventNameTemp)]
        public async Task<ISubscribeAck> ReceivedAFN0CEvent(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>(serverName, receivedMessage.Data);
                return SubscribeAck.Success();
            }

            return SubscribeAck.Fail();
        }


        /// <summary>
        /// 解析AFN0DH
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.SubscriberAFN0DHReceivedEventNameTemp)]
        public async Task<ISubscribeAck> ReceivedAFN0DEvent(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>(serverName, receivedMessage.Data);
                return SubscribeAck.Success();
            }

            return SubscribeAck.Fail();
        }


        /// <summary>
        /// 解析AFN0EH
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.SubscriberAFN0EHReceivedEventNameTemp)]
        public async Task<ISubscribeAck> ReceivedAFN0EEvent(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>(serverName, receivedMessage.Data);
                return SubscribeAck.Success();
            }

            return SubscribeAck.Fail();
        }


        /// <summary>
        /// 解析AFN0HH
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.SubscriberAFN10HReceivedEventNameTemp)]
        public async Task<ISubscribeAck> ReceivedAFN0HEvent(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>(serverName, receivedMessage.Data);
                return SubscribeAck.Success();
            }

            return SubscribeAck.Fail();
        }
    }
}