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

using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.IoTDB.Interface;
using JiShe.CollectBus.IotSystems.MeterReadingRecords;
using JiShe.CollectBus.Kafka.Attributes;
using JiShe.CollectBus.Kafka.Internal;
using Microsoft.Extensions.Logging;
using System;
using System.Threading.Tasks;
using TouchSocket.Sockets;

namespace JiShe.CollectBus.Subscribers
{
    /// <summary>
    /// 定时抄读任务消息消费订阅
    /// </summary>
    public class WorkerSubscriberAppService : CollectBusAppService, IWorkerSubscriberAppService, IKafkaSubscribe
    {
        private readonly ILogger<WorkerSubscriberAppService> _logger;
        private readonly ITcpService _tcpService;
        private readonly IIoTDbProvider _dbProvider;

        /// <summary>
        /// Initializes a new instance of the <see cref="WorkerSubscriberAppService"/> class.
        /// </summary>
        /// <param name="logger">The logger.</param>
        /// <param name="tcpService">The TCP service.</param>
        /// <param name="dbProvider">IoTDB数据驱动</param>
        public WorkerSubscriberAppService(ILogger<WorkerSubscriberAppService> logger,
            ITcpService tcpService,
            IIoTDbProvider dbProvider)
        {
            _logger = logger;
            _tcpService = tcpService;
            _dbProvider = dbProvider;
        }


        #region 电表消息采集 

        /// <summary>
        /// 一分钟定时抄读任务消息消费订阅
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName)]
        public async Task<ISubscribeAck> AmmeterScheduledMeterOneMinuteReadingIssuedEvent(MeterReadingTelemetryPacketInfo receivedMessage)
        {
            _logger.LogInformation("1分钟采集电表数据下行消息消费队列开始处理");
            return await SendMessagesAsync(receivedMessage);

        }

        /// <summary>
        /// 5分钟采集电表数据下行消息消费订阅
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.AmmeterSubscriberWorkerFiveMinuteIssuedEventName)]
        public async Task<ISubscribeAck> AmmeterScheduledMeterFiveMinuteReadingIssuedEvent(MeterReadingTelemetryPacketInfo receivedMessage)
        {
            _logger.LogInformation("5分钟采集电表数据下行消息消费队列开始处理");
            return await SendMessagesAsync(receivedMessage);
        }

        /// <summary>
        /// 15分钟采集电表数据下行消息消费订阅
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.AmmeterSubscriberWorkerFifteenMinuteIssuedEventName)]
        public async Task<ISubscribeAck> AmmeterScheduledMeterFifteenMinuteReadingIssuedEvent(MeterReadingTelemetryPacketInfo receivedMessage)
        {
            _logger.LogInformation("15分钟采集电表数据下行消息消费队列开始处理");
            return await SendMessagesAsync(receivedMessage);
        }

        /// <summary>
        /// 电表自动阀控下行消息消费订阅
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.AmmeterSubscriberWorkerAutoValveControlIssuedEventName)]
        public async Task<ISubscribeAck> AmmeterScheduledAutoValveControl(MeterReadingTelemetryPacketInfo receivedMessage)
        {
            //todo 如果是时段自动阀控，需要检查当前的时间，如果时间在自动阀控时间段内，则发送自动阀控报文，否则不发送，尤其是消息队列阻塞或者延时过长的时候。以免造成生产事故。
            _logger.LogInformation("电表自动阀控下行消息消费队列开始处理");
            return await SendMessagesAsync(receivedMessage);
        }

        /// <summary>
        /// 重试电表下行消息消费队列
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.AmmeterSubscriberWorkerRetryEventName)]
        public async Task<ISubscribeAck> AmmeterScheduledMeterRetryReadingEvent(MeterReadingTelemetryPacketInfo receivedMessage)
        {
            _logger.LogInformation("重试电表数据下行消息消费队列开始处理");
            //return await SendMessagesAsync(receivedMessage);
            try
            {
                var checkResult = _tcpService.ClientExists(receivedMessage.FocusAddress);
                if (checkResult)
                {
                    await _tcpService.SendAsync(receivedMessage.FocusAddress, Convert.FromHexString(receivedMessage.IssuedMessageHexString));
                    receivedMessage.IsSend = true;
                    // 增加·发送次数和重试开始时间
                    receivedMessage.SendNum += 1;
                    receivedMessage.NextSendTime = DateTime.Now.AddHours(1);
                    await _dbProvider.InsertAsync(receivedMessage);
                    // TODO： 第4次的时候会推送到地方预警处理
                }
                // 由于有3次重试机会，故每次消息都会被确认
                return SubscribeAck.Success();
            }
            catch (Exception)
            {
                throw;
            }
        }

        #endregion

        #region 水表消息采集

        /// <summary>
        /// 水表数据下行消息消费订阅
        /// </summary>
        /// <param name="receivedMessage"></param>
        /// <returns></returns>
        [KafkaSubscribe(ProtocolConst.WatermeterSubscriberWorkerAutoReadingIssuedEventName)]
        public async Task<ISubscribeAck> WatermeterSubscriberWorkerAutoReadingIssuedEvent(MeterReadingTelemetryPacketInfo receivedMessage)
        {
            return await SendMessagesAsync(receivedMessage);
        }
        #endregion

        /// <summary>
        /// 设备报文发送
        /// </summary>
        /// <param name="receivedMessage">消息记录</param>
        /// <returns></returns>
        private async Task<ISubscribeAck> SendMessagesAsync(MeterReadingTelemetryPacketInfo receivedMessage)
        {
            try
            {
                var checkResult = _tcpService.ClientExists(receivedMessage.FocusAddress);
                if (checkResult)
                {
                    await _tcpService.SendAsync(receivedMessage.FocusAddress, Convert.FromHexString(receivedMessage.IssuedMessageHexString));

                    receivedMessage.IsSend = true;
                    // 增加·发送次数和重试开始时间
                    receivedMessage.SendNum += 1;
                    receivedMessage.NextSendTime = DateTime.Now.AddHours(1);
                    await _dbProvider.InsertAsync(receivedMessage);

                    return SubscribeAck.Success();
                }
                else
                {
                    return SubscribeAck.Fail();
                }
            }
            catch (Exception)
            {
                throw;
            }
        }
    }
}