using Confluent.Kafka;
using static Confluent.Kafka.ConfigPropertyNames;

namespace YLErp.Helpers
{
    public class KafkaConsumerHelper : IDisposable
    {
        private ConsumerConfig _config;
        private IConsumer<Ignore, string> _consumer;
        private string _topic;
        private bool _enableAutoCommit=true;

        public KafkaConsumerHelper(string groupId, string topic)
        {
           var bootstrapServers = Environment.GetEnvironmentVariable("KafkaConfig_BootstrapServers");
           var enableAutoCommit= Environment.GetEnvironmentVariable("KafkaConfig_EnableAutoCommit");
            if (!string.IsNullOrEmpty(enableAutoCommit))
            {
                _enableAutoCommit = bool.Parse(enableAutoCommit);
            }
            _config = new ConsumerConfig
            {
                BootstrapServers = bootstrapServers,
                AutoOffsetReset = AutoOffsetReset.Earliest,
                AllowAutoCreateTopics = true,
                EnableAutoCommit = _enableAutoCommit,

            };
            if (!string.IsNullOrEmpty(groupId))
            {
                _config.GroupId = groupId;
            }
            _consumer = new ConsumerBuilder<Ignore, string>(_config).Build();
            _topic = topic;
        }

        public void Subscribe(Action<string> messageHandler)
        {
            _consumer.Subscribe(_topic);
            try
            {
                while (true)
                {
                    var message = _consumer.Consume();
                    messageHandler?.Invoke(message.Message.Value);
                    if (!_enableAutoCommit)
                    {
                        _consumer.Commit();
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // 当取消订阅时退出循环
            }
        }

        public void Dispose()
        {
            _consumer.Close();
            _consumer.Dispose();
        }
    }
}