package com.elab.mq.config;

import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.elab.log.ext.CatIdMonitorRule;
import com.elab.mq.listener.AbstractMessageListener;
import com.elab.mq.listener.MessageListenerWrapper;
import com.elab.mq.msg.IMsgProducerFacade;
import com.elab.mq.msg.impl.MsgProducerImpl;
import com.google.common.base.Splitter;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.core.env.Environment;

import java.util.*;

/**
 * rocketMQ消息队列的配置
 *
 * @author ： liukx
 * @time ： 2019/4/19 - 16:41
 */
@Configuration
//@PropertySource(value = {"classpath:application.properties"})
@Import({MessageListenerWrapper.class})
public class RocketMQConfiguration {

    private final String prefix = "elab.mq";

    private Logger logger = LoggerFactory.getLogger(RocketMQConfiguration.class);

    @Bean
    public IMsgProducerFacade producerBean(@Autowired Environment env) {
        IMsgProducerFacade producerBean = new MsgProducerImpl();
        producerBean.setProperties(mqProperties(env));
        producerBean.start();
        return producerBean;
    }

    @Bean
    @ConditionalOnMissingBean(value = {CatIdMonitorRule.class})
    public CatIdMonitorRule topicMonitorRule() {
        return new CatIdMonitorRule();
    }

    @Bean
    @ConditionalOnBean(value = {AbstractMessageListener.class})
    public Consumer consumerBean(@Autowired Environment env, @Autowired MessageListenerWrapper
            messageListenerWrapper, @Autowired List<AbstractMessageListener> messageListeners) {
        Consumer consumer = ONSFactory.createConsumer(mqProperties(env));
        if (messageListeners.size() > 0) {
            Map<String, Set<String>> topicMap = new LinkedHashMap<>();
            for (int i = 0; i < messageListeners.size(); i++) {
                AbstractMessageListener abstractMessageListener = messageListeners.get(i);
                String topic = abstractMessageListener.topic();
                Set<String> tags = topicMap.get(topic);
                if (tags == null) {
                    tags = new HashSet<>();
                    topicMap.put(topic, tags);
                }
                String tag = abstractMessageListener.tag();
                List<String> tagList = Splitter.on("||").trimResults().omitEmptyStrings().splitToList(tag);
                tags.addAll(tagList);
            }
            topicMap.forEach((K, V) -> {
                String tag = StringUtils.join(V.toArray(), " || ");
                consumer.subscribe(K, tag, messageListenerWrapper);
                logger.debug(" MQ 消费者注册 -> topic : {} , tag : {}", K, tag);
            });
            consumer.start();
        }
        return consumer;
    }

    /**
     * 通用的参数属性
     *
     * @param env
     * @return
     */
    public Properties mqProperties(Environment env) {
        Properties properties = new Properties();
        // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.AccessKey, env.getProperty(getPrefix(PropertyKeyConst.AccessKey)));
        // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.SecretKey, env.getProperty(getPrefix(PropertyKeyConst.SecretKey)));
        // 设置 TCP 接入域名，进入控制台的实例管理页面的“获取接入点信息”区域查看
        properties.put(PropertyKeyConst.NAMESRV_ADDR, env.getProperty(getPrefix(PropertyKeyConst.NAMESRV_ADDR)));
        properties.put(PropertyKeyConst.GROUP_ID, groupId(env));
        //设置发送超时时间，单位毫秒
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, env.getProperty(getPrefix(PropertyKeyConst.SendMsgTimeoutMillis),
                "5000"));
        properties.put(PropertyKeyConst.OnsChannel, "ALIYUN");
        // 顺序消息消费者失败进行重试前的等待时间(单位毫秒)
        properties.put(PropertyKeyConst.SuspendTimeMillis, env.getProperty(getPrefix(PropertyKeyConst
                .SuspendTimeMillis), "100"));
        // 消费失败的重试次数
        properties.put(PropertyKeyConst.MaxReconsumeTimes, env.getProperty(getPrefix(PropertyKeyConst
                .MaxReconsumeTimes), "20"));
        return properties;
    }

    /**
     * 默认的配置文件前缀
     *
     * @param val
     * @return
     */
    private String getPrefix(String val) {
        return prefix + "." + val;
    }

    /**
     * 默认的groupId是运行环境+项目名称为一组
     *
     * @param env
     * @return
     */
    private String groupId(Environment env) {
        String defaultGroupId = "GID-" + env.getProperty("spring.profiles.active") + "-" + env.getProperty("spring" +
                ".application.name");
        return env.getProperty(getPrefix(PropertyKeyConst.GROUP_ID), defaultGroupId);
    }


}
