package com.tsi.comm.server.kafka;

import com.tsi.comm.server.config.TsiCvimServerConfig;
import com.tsi.comm.server.config.TsiKafkaProducerConfig;
import com.tsi.comm.server.process.dbms.TsiCvimDbmsProcess;
import com.tsi.comm.server.repository.TsiTpmsManager;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.TopicPartition;
import org.jetbrains.annotations.NotNull;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ConsumerAwareRebalanceListener;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.stereotype.Service;

import java.util.Collection;

@Slf4j
@RequiredArgsConstructor
@Service
public class KafkaConsumerService {

    private final TsiCvimServerConfig config;
    private final TsiKafkaProducerConfig producerConfig;
    private final TsiTpmsManager tpmsManager;
    private final TsiCvimDbmsProcess dbmsProcess;

    private ConcurrentMessageListenerContainer<String, Long> kafkaListenerContainer = null;

    public void start() {

        if (this.kafkaListenerContainer != null) {
            if (!this.kafkaListenerContainer.isRunning()) {
                log.warn("kafkaListenerContainer restart");
                this.kafkaListenerContainer.start();
            }
            return;
        }

        ContainerProperties containerProperties = new ContainerProperties(TsiKafkaProducerConfig.CVIM_PING+this.config.getServerId());
        containerProperties.setGroupId(this.producerConfig.getGroupId());
        containerProperties.setPollTimeout(5000);
        //containerProperties.setAckMode(ContainerProperties.AckMode.MANUAL);
        containerProperties.setMessageListener(new TsiKafkaConsumerWorker(this.tpmsManager, this.dbmsProcess));
        containerProperties.setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() {
            @Override
            public void onPartitionsRevokedBeforeCommit(@NotNull Consumer<?, ?> consumer, @NotNull Collection<TopicPartition> partitions) {
            }
            @Override
            public void onPartitionsRevokedAfterCommit(@NotNull Consumer<?, ?> consumer, @NotNull Collection<TopicPartition> partitions) {
            }
            @Override
            public void onPartitionsAssigned(@NotNull Consumer<?, ?> consumer, @NotNull Collection<TopicPartition> partitions) {
                consumer.seekToEnd(partitions);
            }
        });

        ConsumerFactory<String, Long> consumerFactory = new DefaultKafkaConsumerFactory<>(this.producerConfig.getConsumerPropertiesMap());
        this.kafkaListenerContainer = new ConcurrentMessageListenerContainer<>(consumerFactory, containerProperties);
        this.kafkaListenerContainer.setBeanName("consumer");
        this.kafkaListenerContainer.setConcurrency(1);
        this.kafkaListenerContainer.setErrorHandler((thrownException, data) -> {
            log.error("kafkaListenerContainer error: {}", thrownException.getMessage());
            this.kafkaListenerContainer.stop();
        });

        this.kafkaListenerContainer.start();
    }

    public void shutdown() {
        try {
            //if (this.consumer != null) {
            //    this.consumer.close();;
            //}
            if (this.kafkaListenerContainer != null) {
                this.kafkaListenerContainer.stop();
            }
        }
        catch(Exception ignored) {
            // Ignored exception during shutdown
        }
    }
}
