package com.hdkj.lt.mq.config;

import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.hdkj.lt.core.hussar.SpringContextHolder;
import com.hdkj.lt.core.mq.config.RocketMQProperties;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * rocketmq consumer 配置
 */
@RequiredArgsConstructor
@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties({RocketMQProperties.class})
@ConditionalOnProperty(value = {"lt.mq.rocketmq.enabled"}, havingValue = "true", matchIfMissing = false)
public class RocketMQConsumerConfig {

    private final RocketMQProperties rocketMQProperties;

    @SneakyThrows(ClassNotFoundException.class)
    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public ConsumerBean buildConsumer() {
        ConsumerBean consumerBean = new ConsumerBean();
        Properties properties = new Properties();
        properties.setProperty(PropertyKeyConst.AccessKey, rocketMQProperties.getAccessKey());
        properties.setProperty(PropertyKeyConst.SecretKey, rocketMQProperties.getSecretKey());
        properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, rocketMQProperties.getNameServer());
        properties.setProperty(PropertyKeyConst.GROUP_ID, rocketMQProperties.getGroup());
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, String.valueOf(rocketMQProperties.getConsumer().getThreadNums()));
        properties.setProperty(PropertyKeyConst.MaxReconsumeTimes, String.valueOf(rocketMQProperties.getConsumer().getMaxReconsumeTimes()));
        consumerBean.setProperties(properties);
        Map<Subscription, MessageListener> subscriptionTable = new HashMap<>();
        List<RocketMQProperties.Topic> topics = rocketMQProperties.getTopic();
        for (RocketMQProperties.Topic topic : topics) {
            String topicName = topic.getTopicName();
            Class<?> listenerClass = Class.forName(topic.getListenerClass());
            MessageListener listener = (MessageListener) SpringContextHolder.getBean(listenerClass);
            Subscription subscription = this.getSubscription(topicName);
            subscriptionTable.put(subscription, listener);
        }
        consumerBean.setSubscriptionTable(subscriptionTable);
        return consumerBean;
    }

    /**
     * 获取订阅相关对象
     *
     * @param topic 订阅名称
     * @return 订阅对象
     */
    private Subscription getSubscription(String topic) {
        Subscription subscription = new Subscription();
        subscription.setTopic(topic);
        subscription.setExpression("*");
        return subscription;
    }

}
