package com.tsi.comm.server.kafka;

import com.tsi.app.common.utils.TimeUtils;
import com.tsi.comm.server.mybatis.vo.AbstractDbmsVo;
import com.tsi.comm.server.mybatis.vo.KafkaTransVo;
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.ConsumerRecord;
import org.jetbrains.annotations.NotNull;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.support.Acknowledgment;

import java.util.concurrent.TimeUnit;

@Slf4j
@RequiredArgsConstructor
public class TsiKafkaConsumerWorker implements MessageListener<String, Long> {

    private final TsiTpmsManager tpmsManager;
    private final TsiCvimDbmsProcess dbmsProcess;

    @Override
    public void onMessage(ConsumerRecord<String, Long> consumerRecord) {
        Long sendNanoTime = consumerRecord.value();
        Long recvNanoTime = System.nanoTime();

        KafkaTransVo stat = new KafkaTransVo(AbstractDbmsVo.DBMS_KAFKA_TRANS_HS);
        stat.setHostName(this.tpmsManager.getKafkaTransVo().getHostName());
        stat.setStatus(1);
        if (this.tpmsManager.getKafkaTransVo().getSendNanoTime() == sendNanoTime) {
            stat.setSendTm(this.tpmsManager.getKafkaTransVo().getSendTm());
            stat.setRecvTm(TimeUnit.MICROSECONDS.convert(Math.abs(recvNanoTime - sendNanoTime), TimeUnit.NANOSECONDS));
        }
        else {
            stat.setRecvTm(TimeUnit.MICROSECONDS.convert(Math.abs(recvNanoTime - sendNanoTime), TimeUnit.NANOSECONDS));
            stat.setSendTm(stat.getRecvTm());
            log.info("recv ping success, sendNanoTime miss match: {}, {}", sendNanoTime, TimeUtils.elapsedTimeStr(recvNanoTime - sendNanoTime));
        }
        dbmsProcess.add(stat, (int)Thread.currentThread().getId());
        log.info("recv ping success: {}, {}", sendNanoTime, TimeUtils.elapsedTimeStr(recvNanoTime - sendNanoTime));
    }

    @Override
    public void onMessage(@NotNull ConsumerRecord<String, Long> consumerRecord, Acknowledgment acknowledgment) {
        try {
            Long sendNanoTime = consumerRecord.value();
            Long recvNanoTime = System.nanoTime();
            log.info("recv ping success, ack: {}, {}", sendNanoTime, TimeUtils.elapsedTimeStr(recvNanoTime - sendNanoTime));
            //acknowledgment.acknowledge();
        } catch (Exception e) {
            log.error("KafkaConsumerWorker.onMessage：{}", e.getMessage());
        }
    }

}
