using Confluent.Kafka;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using TelpoPush.WanJiaAn.Worker.Models.Config;

namespace TelpoPush.WanJiaAn.Worker.Service.Mq
{
    public class KafkaService : IKafkaService
    {
        private readonly ConsumerConfig _consumerConfig;
        private readonly IHostEnvironment env;
        private readonly ILogger<KafkaService> logger;
        private readonly ServiceConfig _configService;
        public KafkaService(ILogger<KafkaService> _logger, IHostEnvironment _env, IOptions<ServiceConfig> optConfigService)
        {
            _configService = optConfigService.Value;
            env = _env;
            logger = _logger;

            //_consumerConfig = new ConsumerConfig
            //{
            //    BootstrapServers = _configService.KafkaBootstrapServers,
            //    GroupId = _configService.KafkaGroupId,
            //    EnableAutoCommit = false, // 禁止AutoCommit
            //    Acks = Acks.Leader, // 假设只需要Leader响应即可
            //    AutoOffsetReset = AutoOffsetReset.Earliest,// 从最早的开始消费起
            //    CancellationDelayMaxMs = 1//set CancellationDelayMaxMs
            //};



            _consumerConfig = new ConsumerConfig
            {
                BootstrapServers = _configService.KafkaBootstrapServers,
                SecurityProtocol = SecurityProtocol.SaslPlaintext,
                SaslMechanism = SaslMechanism.Plain,
                GroupId = _configService.KafkaGroupId,
                SaslUsername = _configService.KafkaUserName,
                SaslPassword = _configService.KafkaPassword,
                EnableAutoCommit = false, // 禁止AutoCommit
                AutoOffsetReset = AutoOffsetReset.Earliest, // 从最早的开始消费起
                CancellationDelayMaxMs = 1
            };


        }
        public async Task SubscribeAsync(Action<string, string, Headers> messageFunc, CancellationToken cancellationToken)
        {
            List<string> topics = _configService.KafkaTopics;
            using (var consumer = new ConsumerBuilder<Ignore, string>(_consumerConfig)
                                 .SetErrorHandler((_, e) =>
                                 {
                                     logger.LogError($"Error: {e.Reason}");
                                 })
                                 .SetStatisticsHandler((_, json) =>
                                 {
                                     logger.LogInformation($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} > 消息监听中..");
                                 })
                                 .SetPartitionsAssignedHandler((c, partitions) =>
                                 {
                                     string partitionsStr = string.Join(", ", partitions);
                                     logger.LogInformation($" - 分配的 kafka 分区: {partitionsStr}");
                                 })
                                 .SetPartitionsRevokedHandler((c, partitions) =>
                                 {
                                     string partitionsStr = string.Join(", ", partitions);
                                     logger.LogInformation($" - 回收了 kafka 分区: {partitionsStr}");
                                 })
                                 .Build())
            {

                consumer.Subscribe(topics);
                try
                {
                    while (true)
                    {
                        try
                        {
                            var consumeResult = consumer.Consume(cancellationToken);
                            string topic = consumeResult.Topic;
                            string messageResult = consumeResult.Message.Value;
                            Headers headers = consumeResult.Message.Headers;
                            bool isPartitionEOF = consumeResult.IsPartitionEOF;
                            var partition = consumeResult.Partition;

                            int DataType = -1, AlarmType = -1, OperType = -1;
                            foreach (var item in headers)
                            {
                                if (item.Key == KafkaHeader.DataType)
                                    DataType = BitConverter.ToInt32(item.GetValueBytes(), 0);
                                else if (item.Key == KafkaHeader.AlarmType)
                                    AlarmType = BitConverter.ToInt32(item.GetValueBytes(), 0);
                                else if (item.Key == KafkaHeader.OperType)
                                    OperType = BitConverter.ToInt32(item.GetValueBytes(), 0);
                            }
                            var Headers = new { DataType, AlarmType, OperType };
                            logger.LogInformation($"Consumed  topic '{topic}' , message '{messageResult}' ,  headers '{JsonConvert.SerializeObject(Headers)}', at '{consumeResult?.TopicPartitionOffset}'.");
                            if (isPartitionEOF)
                            {
                                logger.LogInformation($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} 已经到底了：{topic}, partition {partition}, offset {consumeResult?.Offset}.");
                                continue;
                            }
                            if (!string.IsNullOrEmpty(messageResult))
                            {
                                messageFunc(topic, messageResult, headers);
                                try
                                {
                                    consumer.Commit(consumeResult);
                                }
                                catch (KafkaException e)
                                {
                                    logger.LogError($" - {e.Message}.");
                                }
                            }
                            #region 注释
                            //string messageResult = null;
                            //Headers headers = null;
                            //try
                            //{
                            //    messageResult = consumeResult.Message.Value;
                            //    headers = consumeResult.Message.Headers;
                            //}
                            //catch (Exception ex)
                            //{
                            //    var errorMessage = $" - {DateTime.Now:yyyy-MM-dd HH:mm:ss}【Exception 消息反序列化失败，Value：{consumeResult.Message.Value}】 ：{ex.StackTrace?.ToString()}";
                            //    Console.WriteLine(errorMessage);
                            //    logger.LogError(errorMessage);
                            //    messageResult = null;
                            //}
                            //if (!string.IsNullOrEmpty(messageResult)/* && consumeResult.Offset % commitPeriod == 0*/)
                            //{
                            //    string topic = consumeResult.Topic;
                            //    messageFunc(topic, messageResult, headers);

                            //    //try
                            //    //{
                            //    //    consumer.Commit(consumeResult);
                            //    //}
                            //    //catch (KafkaException e)
                            //    //{
                            //    //    Console.WriteLine(e.Message);
                            //    //}
                            //}
                            #endregion
                        }
                        catch (ConsumeException e)
                        {
                            logger.LogError($"Consume error: {e.Error.Reason}");
                        }
                    }
                }
                catch (OperationCanceledException)
                {
                    logger.LogError("Closing consumer.");
                    consumer.Close();
                }
            }
            await Task.CompletedTask;
        }
    }
}