package com.java110.log.kafka;

import com.java110.core.base.controller.BaseController;
import com.java110.log.smo.ILogServiceSMO;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;

/**
 * kafka侦听
 * 接受到的信息报文格式为：
 * {
 *     "transactionId":"交易流水号",
 *     "dataFlowId":"上下文对象",
 *     "ip":"ip",
 *     "port":"端口",
 *     "srcIp":"调用方IP",
 *     "srcPort":"调用方端口",
 *     "appId":"应用ID",
 *     "userId":"用户ID",
 *     "serviceCode":"服务编码",
 *     "serviceName":"服务名称",
 *     "timestamp":"时间储",
 *     "logStatus":"记录状态",
 *     "requestMessage":"请求信息",
 *     "responseMessage":"返回信息"
 * }
 * Created by wuxw on 2018/4/15.
 */
public class LogServiceKafka extends BaseController {

    private final static Logger logger = LoggerFactory.getLogger(LogServiceKafka.class);


    @Autowired
    private ILogServiceSMO logServiceSMOImpl;

    @KafkaListener(topics = {"LOG"})
    public void listen(ConsumerRecord<?, ?> record) {
        logger.info("LogServiceKafka receive message: {}", record.value().toString());
        String logMessage = record.value().toString();
        logServiceSMOImpl.saveLogMessage(logMessage);
    }



    public ILogServiceSMO getLogServiceSMOImpl() {
        return logServiceSMOImpl;
    }

    public void setLogServiceSMOImpl(ILogServiceSMO logServiceSMOImpl) {
        this.logServiceSMOImpl = logServiceSMOImpl;
    }
}