package cn.freemud.amp.config;

import cn.freemud.constant.RocketMQConst;
import com.alibaba.fastjson.support.spring.messaging.MappingFastJsonMessageConverter;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;

@Configuration
public class RocketMQConfig {

    @Resource
    private RocketMQProperties rocketMQProperties;

    @Bean(name = "deliveryRocketMqTemplate", destroyMethod = "destroy")
    public RocketMQTemplate rocketMQTemplate() {
        DefaultMQProducer defaultMQProducer = createMQProducer(RocketMQConst.ORDER_THIRD_PUSH_PRODUCER_GROUP);
        RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
        rocketMQTemplate.setProducer(defaultMQProducer);
        rocketMQTemplate.setMessageConverter(new MappingFastJsonMessageConverter());
        rocketMQTemplate.setMessageQueueSelector(new SelectMessageQueueByHash());   // MessageQueueSelector, default SelectMessageQueueByHash
        return rocketMQTemplate;
    }

    /**
     * @return
     */
    public DefaultMQProducer createMQProducer(String producerGroupName) {
        RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer();
        String nameServer = rocketMQProperties.getNameServer();
        String groupName = StringUtils.isEmpty(producerGroupName) ? producerConfig.getGroup() : producerGroupName;
        Assert.hasText(nameServer, "rocketmq.name-server must not be null");
        Assert.hasText(groupName, "rocketmq.producer.group must not be null");

        String accessChannel = rocketMQProperties.getAccessChannel();
        String accessKey = rocketMQProperties.getProducer().getAccessKey();
        String secretKey = rocketMQProperties.getProducer().getSecretKey();

        DefaultMQProducer producer = null;
        // 密码模式
        if (!StringUtils.isEmpty(accessKey) && !StringUtils.isEmpty(secretKey)) {
            producer = new DefaultMQProducer(groupName, new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))
                    , rocketMQProperties.getProducer().isEnableMsgTrace()
                    , rocketMQProperties.getProducer().getCustomizedTraceTopic());
            producer.setVipChannelEnabled(false);
        } else {
            producer = new DefaultMQProducer(groupName
                    , rocketMQProperties.getProducer().isEnableMsgTrace()
                    , rocketMQProperties.getProducer().getCustomizedTraceTopic());
        }

        producer.setNamesrvAddr(nameServer);

        if (!StringUtils.isEmpty(accessChannel)) {
            producer.setAccessChannel(AccessChannel.valueOf(accessChannel));
        }

        producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
        producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
        producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
        producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
        producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
        producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
        producer.setPollNameServerInterval(rocketMQProperties.getPollNameServerInteval());
        producer.setHeartbeatBrokerInterval(rocketMQProperties.getHeartbeatBrokerInterval());
        producer.setPersistConsumerOffsetInterval(rocketMQProperties.getPersistConsumerOffsetInterval());

        return producer;
    }

}
