package com.mvc.iot.aep;

import com.alibaba.fastjson.JSON;
import com.ctiot.aep.mqmsgpush.sdk.IMsgConsumer;
import com.ctiot.aep.mqmsgpush.sdk.IMsgListener;
import com.ctiot.aep.mqmsgpush.sdk.MqMsgConsumer;
import com.mvc.iot.IotStoreBlockingQueue;
import com.mvc.iot.aep.config.AepConfig;
import com.mvc.model.aep.AepRootModel;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 电信AEP平台接收器
 */
@Component
public class AepMqListener implements Runnable {
    private static final Logger logger = LoggerFactory.getLogger(AepMqListener.class);

    /**
     * 是否启用AEP消息监听器(防止HTTP监听器与MQ监听器同时启动)
     */
    public static boolean enabledAepMqListener = false;

    @Resource
    private AepConfig aepConfig;

    @Override
    public void run() {
        initAepMq();
    }

    public void initAepMq() {
        String server = aepConfig.getMqServer(); //消息服务地址请在控制台-消息流转-目的地管理-MQ页面中查看
        String tenantId = aepConfig.getMqTenantId();//租户ID
        String token = aepConfig.getMqToken();//身份认证token串
        String certFilePath = ""; //直接填空字符串，CA证书，JDK已经内置相关根证书，无需指定

        if(StringUtils.isEmpty(server)||StringUtils.isEmpty(tenantId)||StringUtils.isEmpty(token)){
            logger.error("AEP-MQ:参数配置不完整，不启用MQ接收");
            return;
        }

        //创建消息接收Listener
        IMsgListener msgListener = msg -> {
            //接收消息
            logger.info("[AEP]MQ消息:{}",msg);
            //消息处理...
            //为了提高效率，建议对消息进行异步处理（使用其它线程、发送到Kafka等）
            try {
                AepRootModel model = JSON.parseObject(msg, AepRootModel.class);
                IotStoreBlockingQueue.aepCacheQueue.put(model);
            } catch (Exception exception) {
                exception.printStackTrace();
            }
        };

        //创建消息接收类
        IMsgConsumer consumer = new MqMsgConsumer();
        try {
            //初始化
            /**
             * @param server  消息服务server地址
             * @param tenantId 租户Id
             * @param token    用户认证token
             * @param certFilePath 证书文件路径
             * @param topicNames   主题名列表，如果该列表为空或null，则自动消费该租户所有主题消息
             * @param msgListener 消息接收者
             * @return 是否初始化成功
             */
            boolean initSuccess = consumer.init(server, tenantId, token, certFilePath, null, msgListener);

            //开始接收消息
            consumer.start();

            enabledAepMqListener = initSuccess;

            //程序退出时，停止接收、销毁
            //consumer.stop();
            //consumer.destroy();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
