package com.hdkj.lt.core.mq.client;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.bean.ProducerBean;
import com.hdkj.lt.core.mq.config.RocketMQProperties;
import com.hdkj.lt.core.mq.model.RocketMQMessageEntity;
import com.hdkj.lt.core.plugins.thread.MultiThreadExecuteService;
import com.hdkj.lt.utils.cache.RedisRepository;
import lombok.RequiredArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.stereotype.Component;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;

/**
 * @Description:
 * @Author: wenhao
 * @Create: 2024/06/06 9:09
 * @Version: 1.0
 */
@Component
@RequiredArgsConstructor
@EnableConfigurationProperties({RocketMQProperties.class})
public class RocketMQTemplate {

    @Autowired(required = false)
    @Qualifier("buildProducer")
    private ProducerBean producerBean; //如果没有开启MQ，则不需要加载此类

    private final RedisRepository redisRepository;

    private final RocketMQProperties rocketMQProperties;

    private final MultiThreadExecuteService multiThreadExecuteService;


    public void sendMessage(RocketMQMessageEntity messageEntity) {
        if (rocketMQProperties.getEnabled()) {
            Message message = this.buildMessage(messageEntity);
            producerBean.send(message);
            return;
        }
        redisRepository.leftPush(messageEntity.getCacheKey(), messageEntity.getBody());
        this.processRedisMessage(messageEntity);
    }

    public void sendOnewayMessage(RocketMQMessageEntity messageEntity) {
        if (rocketMQProperties.getEnabled()) {
            Message message = this.buildMessage(messageEntity);
            producerBean.sendOneway(message);
            return;
        }
        redisRepository.leftPush(messageEntity.getCacheKey(), messageEntity.getBody());
        this.processRedisMessage(messageEntity);
    }

    /**
     * 发送异步消息，如果开启了MQ，则交给MQ Consumer处理，否则交给Redis队列处理，前者适用于云上环境，后者适用于研发环境
     *
     * @param messageEntity
     */
    public void sendAsyncMessage(RocketMQMessageEntity messageEntity) {
        if (rocketMQProperties.getEnabled()) {
            Message message = this.buildMessage(messageEntity);
            producerBean.sendAsync(message, messageEntity.getSendCallback());
            return;
        }
        redisRepository.leftPush(messageEntity.getCacheKey(), messageEntity.getBody());
        this.processRedisMessage(messageEntity);
    }

    private Message buildMessage(RocketMQMessageEntity messageEntity) {
        Message message = new Message();
        Optional<RocketMQProperties.Topic> optional = rocketMQProperties.getTopic().stream().filter(e -> StringUtils.equals(e.getTopicId(), messageEntity.getTopicId())).findFirst();
        optional.ifPresent(e -> message.setTopic(e.getTopicName()));
        message.setTag(messageEntity.getTag());
        message.setBody(JSON.toJSONString(messageEntity.getBody()).getBytes(StandardCharsets.UTF_8));
        return message;
    }

    private void processRedisMessage(RocketMQMessageEntity messageEntity) {
        CompletableFuture.runAsync(() -> {
            RocketMQMessageEntity msgEntity = BeanUtil.copyProperties(messageEntity, RocketMQMessageEntity.class);
            Object msgBody = redisRepository.rightPop(msgEntity.getCacheKey());
            if (Objects.nonNull(msgBody)) {
                List<RocketMQProperties.Topic> topicList = rocketMQProperties.getTopic();
                for (RocketMQProperties.Topic topic : topicList) {
                    if (StringUtils.equals(topic.getTopicId(), msgEntity.getTopicId())) {
                        try {
                            // 加载类
                            Class<?> clazz = Class.forName(topic.getListenerClass());
                            // 获取所有方法
                            Method[] methods = clazz.getMethods();
                            for (Method method : methods) {
                                if (StringUtils.equals(method.getName(), "consume")) {
                                    // 执行`consume`方法
                                    Message message = this.buildMessage(msgEntity);
                                    method.invoke(message, msgEntity.getSendCallback());
                                }
                            }
                        } catch (ClassNotFoundException | IllegalAccessException | InvocationTargetException ex) {
                            ex.printStackTrace();
                        }
                    }
                }
            }
        }, multiThreadExecuteService.getExecutor());
    }
}
