package com.bcxin.message.common.utils;

import com.alibaba.fastjson.JSON;
import com.bcxin.message.dtos.kafka.message.MsgGroupCreateKafkaMessage;
import com.bcxin.message.dtos.kafka.message.MsgSendRequestKafkaMessage;
import com.bcxin.message.dtos.kafka.record.MsgGroupCreateKafkaRecord;
import com.bcxin.message.dtos.kafka.record.MsgSendRequestKafkaRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;

/**
 * 提取kafkare的record中的信息
 */
public class KafkaMessageUtil {

    /**
     * 消息发送请求表信息转换
     */
    public static MsgSendRequestKafkaMessage transferKafkaMsgSendRequestRecord2Message(ConsumerRecord<String, String> consumerRecord) {
        MsgSendRequestKafkaMessage message = new MsgSendRequestKafkaMessage();
        MsgSendRequestKafkaRecord record = JSON.parseObject(consumerRecord.value(), MsgSendRequestKafkaRecord.class);
        MsgSendRequestKafkaRecord.MsgSendRequestRecord before = record.getBefore();
        MsgSendRequestKafkaRecord.MsgSendRequestRecord after = record.getAfter();

        if(after != null) {
            message.setId(after.getId());
            message.setRequestId(after.getRequest_id());
            message.setAppid(after.getAppid());
            message.setSender(after.getSender());
            message.setTargetType(after.getTarget_type());
            message.setTargetIds(after.getTarget_ids());
            message.setMessageType(after.getMessage_type());
            message.setMessageContent(after.getMessage_content());
            message.setStatus(after.getStatus());
        }
        return message;
    }

    /**
     * 群组创建表信息转换
     */
    public static MsgGroupCreateKafkaMessage transferKafkaMsgGroupCreateRecord2Message(ConsumerRecord<String, String> consumerRecord) {
        MsgGroupCreateKafkaMessage message = new MsgGroupCreateKafkaMessage();
        MsgGroupCreateKafkaRecord record = JSON.parseObject(consumerRecord.value(), MsgGroupCreateKafkaRecord.class);
        MsgGroupCreateKafkaRecord.MsgGroupCreateRecord before = record.getBefore();
        MsgGroupCreateKafkaRecord.MsgGroupCreateRecord after = record.getAfter();

        if(after != null) {
            message.setId(after.getId());
            message.setThirdId(after.getThird_id());
            message.setGroupInfo(after.getGroup_info());
            message.setMembers(after.getMembers());
            message.setOperator(after.getOperator());
            message.setStatus(after.getStatus());
            message.setGroupId(after.getGroup_id());
        }
        return message;
    }
}
