package com.bcxin.message.consumer;

import com.alibaba.fastjson.JSONObject;
import com.bcxin.message.common.CommonConstant;
import com.bcxin.message.common.emus.MsgSendRequestStatus;
import com.bcxin.message.common.utils.KafkaMessageUtil;
import com.bcxin.message.dtos.kafka.message.MsgSendRequestKafkaMessage;
import com.bcxin.message.service.messagecenter.MsgSendRequestService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * 驻勤点kafka监听器
 */
@Slf4j
@Component
public class MsgSendRequestKafkaConsumer {

    @Autowired
    private MsgSendRequestService msgSendRequestService;

    @KafkaListener(id = "bcx_message_center_listener_msg_send_request",
            topics = CommonConstant.MSG_SEND_REQUEST_KAFKA_TOPIC,
            groupId = "bcx_message_center_listener_msg_send_request")
    public void bcxMessageCenterListenerMsgSendRequest(ConsumerRecord<String, String> record, Acknowledgment ack) {
        MsgSendRequestKafkaMessage kafkaMessage = KafkaMessageUtil.transferKafkaMsgSendRequestRecord2Message(record);
        log.info("kafka监听：messagecenter.msg_send_request表，id：{}", kafkaMessage.getId());
        try {
            if (kafkaMessage.getId() != null
                    && (kafkaMessage.getStatus() == null
                    || kafkaMessage.getStatus() == MsgSendRequestStatus.NOT_SEND.getCode())) {
                log.info("调用发送消息方法");
                msgSendRequestService.sendMessage(kafkaMessage);
            }
        } catch (Exception e) {
            log.error("kafka监听：messagecenter.msg_send_request表，异常：{}", e.getMessage(), e);
        }
        ack.acknowledge();
    }
}
