package com.walker.push.rocketmq;

import com.walker.infrastructure.utils.JsonUtils;
import com.walker.infrastructure.utils.StringUtils;
import com.walker.queue.MqBaseMessage;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

/**
 * 加强过的消息模板对象，提供了指定环境的主题参数。
 * <p>目前还没有实现异步发送，后续要加上。</p>
 * @author 时克英
 * @date 2023-09-19
 */
public class RocketMQEnhanceTemplate {

    protected final transient Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * 发送同步消息。
     * @param topic
     * @param message
     * @param delayLevel
     * @return
     * @param <T>
     * @date 2023-09-26
     */
    public <T extends MqBaseMessage> SendResult sendAndWait(String topic, T message, int delayLevel){
        Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
        return template.syncSend(topic, sendMessage, 3000, delayLevel);
    }

    public <T extends MqBaseMessage> void sendAsync(String topic, T message, SendCallback callback){
        this.sendAsync(topic, message, callback, 3000, 0);
    }

    /**
     * 发送异步消息，聊天对象使用该方法。
     * @param topic 队列主题名称，业务会根据主机拼接最终名称
     * @param message 消息对象
     * @param callback 通知回调
     * @param timeout 超时时间（毫秒）
     * @param delayLevel 延时级别，暂时没用，默认：0
     * @param <T>
     * @date 2023-09-26
     */
    public <T extends MqBaseMessage> void sendAsync(String topic
            , T message, SendCallback callback, long timeout, int delayLevel){
        if(StringUtils.isEmpty(topic)){
            throw new IllegalArgumentException("无法发送消息，topic为空");
        }
        Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
//        this.template.asyncSend(topic, sendMessage, callback);
        if(delayLevel > 0){
            this.template.asyncSend(topic, sendMessage, callback, timeout, delayLevel);
        } else {
            this.template.asyncSend(topic, sendMessage, callback);
        }
    }

    //~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    //~ 以上为新方法。2023-09-26
    //~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

    /**
     * 根据系统上下文自动构建隔离后的topic
     * 构建目的地
     */
    public String buildDestination(String topic, String tag) {
        topic = reBuildTopic(topic);
        return topic + StringUtils.SEPARATOR_COLON + tag;
    }

    /**
     * 根据环境重新隔离topic
     * @param topic 原始topic
     */
    private String reBuildTopic(String topic) {
        if(this.enabledIsolation && StringUtils.hasText(this.environmentName)){
            return topic + StringUtils.STRING_UNDERLINE + this.environmentName;
        }
        return topic;
    }

    /**
     * 发送同步消息
     */
    public <T extends MqBaseMessage> SendResult send(String topic, String tag, T message) {
        // 注意分隔符
        return send(buildDestination(topic,tag), message);
    }

    public <T extends MqBaseMessage> SendResult send(String destination, T message) {
        // 设置业务键，此处根据公共的参数进行处理
        // 更多的其它基础业务处理...
        Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
        SendResult sendResult = template.syncSend(destination, sendMessage);
        // 此处为了方便查看给日志转了json，根据选择选择日志记录方式，例如ELK采集
        try {
            logger.debug("[{}]同步消息[{}]发送结果[{}]", destination, JsonUtils.objectToJsonString(message), JsonUtils.objectToJsonString(sendResult));
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return sendResult;
    }

    /**
     * 发送延迟消息
     */
    public <T extends MqBaseMessage> SendResult send(String topic, String tag, T message, int delayLevel) {
        return send(buildDestination(topic,tag), message, delayLevel);
    }

    public <T extends MqBaseMessage> SendResult send(String destination, T message, int delayLevel) {
        Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
        SendResult sendResult = template.syncSend(destination, sendMessage, 3000, delayLevel);
        try {
            logger.debug("[{}]延迟等级[{}]消息[{}]发送结果[{}]", destination, delayLevel, JsonUtils.objectToJsonString(message), JsonUtils.objectToJsonString(sendResult));
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return sendResult;
    }

    public RocketMQTemplate getTemplate() {
        return template;
    }

    public void setTemplate(RocketMQTemplate template) {
        this.template = template;
    }

    public void setEnabledIsolation(boolean enabledIsolation) {
        this.enabledIsolation = enabledIsolation;
    }

    public void setEnvironmentName(String environmentName) {
        this.environmentName = environmentName;
    }
    private boolean enabledIsolation = true;

    private String environmentName = StringUtils.EMPTY_STRING;

    private RocketMQTemplate template;
}