package com.tofly.mq.config;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

/**
 * @author tuonbed
 * @date 2020/6/1  10:32
 */

@Component
@Slf4j
public class KafkaRroducer {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    public boolean sendMsg(Object object){
        //Class oc=object.getClass();
        Class objectClass=object.getClass();
        boolean annotationPresent = objectClass.isAnnotationPresent(ToflyKafkaPojo.class);
        if(!annotationPresent){
            log.error("在实体信息中未发现@ToflyKafkaPojo注解");
            return false;
        }
        ToflyKafkaPojo toflyKafkaPojo= AnnotationUtils.findAnnotation(objectClass,ToflyKafkaPojo.class);
        String topic = toflyKafkaPojo.topic();
        Assert.notNull(topic, "topic is null");
        String key = toflyKafkaPojo.key();
        return sendMsg(topic,JSONObject.toJSONString(object),key);
    }


    public boolean sendMsg(String topic,String msg,String key){
        if(StringUtils.isEmpty(key)){
            kafkaTemplate.send(topic,msg);
        }else{
            kafkaTemplate.send(topic,key,msg);
        }
        return true;
    }

    public boolean sendMsg(String topic,String msg){
        return sendMsg(topic,msg,null);
    }
}
