/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.rocketmq.spring.core;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Map;
import java.util.Objects;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.RequestCallback;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.converter.SmartMessageConverter;
import org.springframework.messaging.core.AbstractMessageSendingTemplate;
import org.springframework.messaging.core.MessagePostProcessor;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeTypeUtils;

@SuppressWarnings({"WeakerAccess", "unused"})
public class RocketMQTemplate extends AbstractMessageSendingTemplate<String> implements InitializingBean, DisposableBean {
    private static final Logger log = LoggerFactory.getLogger(RocketMQTemplate.class);

    private DefaultMQProducer producer;

    private String charset = "UTF-8";

    private MessageQueueSelector messageQueueSelector = new SelectMessageQueueByHash();

    private RocketMQMessageConverter rocketMQMessageConverter = new RocketMQMessageConverter();

    public DefaultMQProducer getProducer() {
        return producer;
    }

    public void setProducer(DefaultMQProducer producer) {
        this.producer = producer;
    }

    public String getCharset() {
        return charset;
    }

    public void setCharset(String charset) {
        this.charset = charset;
    }

    public MessageQueueSelector getMessageQueueSelector() {
        return messageQueueSelector;
    }

    public void setMessageQueueSelector(MessageQueueSelector messageQueueSelector) {
        this.messageQueueSelector = messageQueueSelector;
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message} the message to be sent.
     * @param type The type of T
     * @return
     */
    public <T> T sendAndReceive(String destination, Message<?> message, Type type) {
        return sendAndReceive(destination, message, type, null, producer.getSendMsgTimeout(), 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param payload the payload to be sent.
     * @param type The type of T
     * @return
     */
    public <T> T sendAndReceive(String destination, Object payload, Type type) {
        return sendAndReceive(destination, payload, type, null, producer.getSendMsgTimeout(), 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message} the message to be sent.
     * @param type The type of T
     * @param timeout send timeout in millis
     * @return
     */
    public <T> T sendAndReceive(String destination, Message<?> message, Type type, long timeout) {
        return sendAndReceive(destination, message, type, null, timeout, 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param payload the payload to be sent.
     * @param type The type of T
     * @param timeout send timeout in millis
     * @return
     */
    public <T> T sendAndReceive(String destination, Object payload, Type type, long timeout) {
        return sendAndReceive(destination, payload, type, null, timeout, 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message} the message to be sent.
     * @param type The type of T
     * @param timeout send timeout in millis
     * @param delayLevel message delay level(0 means no delay)
     * @return
     */
    public <T> T sendAndReceive(String destination, Message<?> message, Type type, long timeout, int delayLevel) {
        return sendAndReceive(destination, message, type, null, timeout, delayLevel);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param payload the payload to be sent.
     * @param type The type of T
     * @param timeout send timeout in millis
     * @param delayLevel message delay level(0 means no delay)
     * @return
     */
    public <T> T sendAndReceive(String destination, Object payload, Type type, long timeout, int delayLevel) {
        return sendAndReceive(destination, payload, type, null, timeout, delayLevel);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message} the message to be sent.
     * @param type The type of T
     * @param hashKey needed when sending message orderly
     * @return
     */
    public <T> T sendAndReceive(String destination, Message<?> message, Type type, String hashKey) {
        return sendAndReceive(destination, message, type, hashKey, producer.getSendMsgTimeout(), 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param payload the payload to be sent.
     * @param type The type of T
     * @param hashKey needed when sending message orderly
     * @return
     */
    public <T> T sendAndReceive(String destination, Object payload, Type type, String hashKey) {
        return sendAndReceive(destination, payload, type, hashKey, producer.getSendMsgTimeout(), 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message} the message to be sent.
     * @param type The type of T
     * @param hashKey needed when sending message orderly
     * @param timeout send timeout in millis
     * @return
     */
    public <T> T sendAndReceive(String destination, Message<?> message, Type type, String hashKey, long timeout) {
        return sendAndReceive(destination, message, type, hashKey, timeout, 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param payload the payload to be sent.
     * @param type The type of T
     * @param hashKey
     * @return
     */
    public <T> T sendAndReceive(String destination, Object payload, Type type, String hashKey, long timeout) {
        return sendAndReceive(destination, payload, type, hashKey, timeout, 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message} the message to be sent.
     * @param type The type that receive
     * @param hashKey needed when sending message orderly
     * @param timeout send timeout in millis
     * @param delayLevel message delay level(0 means no delay)
     * @return
     */
    public <T> T sendAndReceive(String destination, Message<?> message, Type type, String hashKey,
        long timeout, int delayLevel) {
        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
            log.error("send request message failed. destination:{}, message is null ", destination);
            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
        }

        try {
            org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
            if (delayLevel > 0) {
                rocketMsg.setDelayTimeLevel(delayLevel);
            }
            MessageExt replyMessage;

            if (Objects.isNull(hashKey) || hashKey.isEmpty()) {
                replyMessage = (MessageExt) producer.request(rocketMsg, timeout);
            } else {
                replyMessage = (MessageExt) producer.request(rocketMsg, messageQueueSelector, hashKey, timeout);
            }
            return replyMessage != null ? (T) doConvertMessage(replyMessage, type) : null;
        } catch (Exception e) {
            log.error("send request message failed. destination:{}, message:{} ", destination, message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param payload the payload to be sent.
     * @param type The type that receive
     * @param hashKey needed when sending message orderly
     * @param timeout send timeout in millis
     * @param delayLevel message delay level(0 means no delay)
     * @return
     */
    public <T> T sendAndReceive(String destination, Object payload, Type type, String hashKey,
        long timeout, int delayLevel) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        return sendAndReceive(destination, message, type, hashKey, timeout, delayLevel);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message} the message to be sent.
     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
     * @return
     */
    public void sendAndReceive(String destination, Message<?> message,
        RocketMQLocalRequestCallback rocketMQLocalRequestCallback) {
        sendAndReceive(destination, message, rocketMQLocalRequestCallback, null, producer.getSendMsgTimeout(), 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param payload the payload to be sent.
     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
     * @return
     */
    public void sendAndReceive(String destination, Object payload,
        RocketMQLocalRequestCallback rocketMQLocalRequestCallback) {
        sendAndReceive(destination, payload, rocketMQLocalRequestCallback, null, producer.getSendMsgTimeout(), 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message} the message to be sent.
     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
     * @param timeout send timeout in millis
     * @return
     */
    public void sendAndReceive(String destination, Message<?> message,
        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, long timeout) {
        sendAndReceive(destination, message, rocketMQLocalRequestCallback, null, timeout, 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param payload the payload to be sent.
     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
     * @param timeout send timeout in millis
     * @return
     */
    public void sendAndReceive(String destination, Object payload,
        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, long timeout) {
        sendAndReceive(destination, payload, rocketMQLocalRequestCallback, null, timeout, 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message} the message to be sent.
     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
     * @param timeout send timeout in millis
     * @param delayLevel message delay level(0 means no delay)
     * @return
     */
    public void sendAndReceive(String destination, Message<?> message,
        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, long timeout, int delayLevel) {
        sendAndReceive(destination, message, rocketMQLocalRequestCallback, null, timeout, delayLevel);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param payload the payload to be sent.
     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
     * @param hashKey needed when sending message orderly
     * @return
     */
    public void sendAndReceive(String destination, Object payload,
        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey) {
        sendAndReceive(destination, payload, rocketMQLocalRequestCallback, hashKey, producer.getSendMsgTimeout(), 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message} the message to be sent.
     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
     * @param hashKey needed when sending message orderly
     * @param timeout send timeout in millis
     * @return
     */
    public void sendAndReceive(String destination, Message<?> message,
        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey, long timeout) {
        sendAndReceive(destination, message, rocketMQLocalRequestCallback, hashKey, timeout, 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param payload the payload to be sent.
     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
     * @param hashKey needed when sending message orderly
     * @param timeout send timeout in millis
     * @return
     */
    public void sendAndReceive(String destination, Object payload,
        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey, long timeout) {
        sendAndReceive(destination, payload, rocketMQLocalRequestCallback, hashKey, timeout, 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message} the message to be sent.
     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
     * @param hashKey needed when sending message orderly
     * @return
     */
    public void sendAndReceive(String destination, Message<?> message,
        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey) {
        sendAndReceive(destination, message, rocketMQLocalRequestCallback, hashKey, producer.getSendMsgTimeout(), 0);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param payload the payload to be sent.
     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
     * @param timeout send timeout in millis
     * @param delayLevel message delay level(0 means no delay)
     * @return
     */
    public void sendAndReceive(String destination, Object payload,
        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, long timeout, int delayLevel) {
        sendAndReceive(destination, payload, rocketMQLocalRequestCallback, null, timeout, delayLevel);
    }

    /**
     * @param destination formats: `topicName:tags`
     * @param payload the payload to be sent.
     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
     * @param hashKey needed when sending message orderly
     * @param timeout send timeout in millis
     * @param delayLevel message delay level(0 means no delay)
     * @return
     */
    public void sendAndReceive(String destination, Object payload,
        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey, long timeout, int delayLevel) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        sendAndReceive(destination, message, rocketMQLocalRequestCallback, hashKey, timeout, delayLevel);
    }

    /**
     * Send request message in asynchronous mode. </p> This method returns immediately. On receiving reply message,
     * <code>rocketMQLocalRequestCallback</code> will be executed. </p>
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message} the message to be sent.
     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
     * @param hashKey needed when sending message orderly
     * @param timeout send timeout in millis
     * @param delayLevel message delay level(0 means no delay)
     * @return
     */
    public void sendAndReceive(String destination, Message<?> message,
        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey, long timeout, int delayLevel) {
        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
            log.error("send request message failed. destination:{}, message is null ", destination);
            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
        }

        try {
            org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
            if (delayLevel > 0) {
                rocketMsg.setDelayTimeLevel(delayLevel);
            }
            if (timeout <= 0) {
                timeout = producer.getSendMsgTimeout();
            }
            RequestCallback requestCallback = null;
            if (rocketMQLocalRequestCallback != null) {
                requestCallback = new RequestCallback() {
                    @Override public void onSuccess(org.apache.rocketmq.common.message.Message message) {
                        rocketMQLocalRequestCallback.onSuccess(doConvertMessage((MessageExt) message, getMessageType(rocketMQLocalRequestCallback)));
                    }

                    @Override public void onException(Throwable e) {
                        rocketMQLocalRequestCallback.onException(e);
                    }
                };
            }
            if (Objects.isNull(hashKey) || hashKey.isEmpty()) {
                producer.request(rocketMsg, requestCallback, timeout);
            } else {
                producer.request(rocketMsg, messageQueueSelector, hashKey, requestCallback, timeout);
            }
        } catch (
            Exception e) {
            log.error("send request message failed. destination:{}, message:{} ", destination, message);
            throw new MessagingException(e.getMessage(), e);
        }

    }

    /**
     * <p> Send message in synchronous mode. This method returns only when the sending procedure totally completes.
     * Reliable synchronous transmission is used in extensive scenes, such as important notification messages, SMS
     * notification, SMS marketing system, etc.. </p>
     * <p>
     * <strong>Warn:</strong> this method has internal retry-mechanism, that is, internal implementation will retry
     * {@link DefaultMQProducer#getRetryTimesWhenSendFailed} times before claiming failure. As a result, multiple
     * messages may potentially delivered to broker(s). It's up to the application developers to resolve potential
     * duplication issue.
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @return {@link SendResult}
     */
    public SendResult syncSend(String destination, Message<?> message) {
        return syncSend(destination, message, producer.getSendMsgTimeout());
    }

    /**
     * Same to {@link #syncSend(String, Message)} with send timeout specified in addition.
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @param timeout send timeout with millis
     * @return {@link SendResult}
     */
    public SendResult syncSend(String destination, Message<?> message, long timeout) {
        return syncSend(destination, message, timeout, 0);
    }

    /**
     * syncSend batch messages in a given timeout.
     *
     * @param destination formats: `topicName:tags`
     * @param messages Collection of {@link org.springframework.messaging.Message}
     * @param timeout send timeout with millis
     * @return {@link SendResult}
     */
    public <T extends Message> SendResult syncSend(String destination, Collection<T> messages, long timeout) {
        if (Objects.isNull(messages) || messages.size() == 0) {
            log.error("syncSend with batch failed. destination:{}, messages is empty ", destination);
            throw new IllegalArgumentException("`messages` can not be empty");
        }

        try {
            long now = System.currentTimeMillis();
            Collection<org.apache.rocketmq.common.message.Message> rmqMsgs = new ArrayList<>();
            for (Message msg : messages) {
                if (Objects.isNull(msg) || Objects.isNull(msg.getPayload())) {
                    log.warn("Found a message empty in the batch, skip it");
                    continue;
                }
                rmqMsgs.add(this.createRocketMqMessage(destination, msg));
            }

            SendResult sendResult = producer.send(rmqMsgs, timeout);
            long costTime = System.currentTimeMillis() - now;
            if (log.isDebugEnabled()) {
                log.debug("send messages cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());
            }
            return sendResult;
        } catch (Exception e) {
            log.error("syncSend with batch failed. destination:{}, messages.size:{} ", destination, messages.size());
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * Same to {@link #syncSend(String, Message)} with send timeout specified in addition.
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @param timeout send timeout with millis
     * @param delayLevel level for the delay message
     * @return {@link SendResult}
     */
    public SendResult syncSend(String destination, Message<?> message, long timeout, int delayLevel) {
        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
            log.error("syncSend failed. destination:{}, message is null ", destination);
            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
        }
        try {
            long now = System.currentTimeMillis();
            org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
            if (delayLevel > 0) {
                rocketMsg.setDelayTimeLevel(delayLevel);
            }
            SendResult sendResult = producer.send(rocketMsg, timeout);
            long costTime = System.currentTimeMillis() - now;
            if (log.isDebugEnabled()) {
                log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());
            }
            return sendResult;
        } catch (Exception e) {
            log.error("syncSend failed. destination:{}, message:{} ", destination, message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * Same to {@link #syncSend(String, Message)}.
     *
     * @param destination formats: `topicName:tags`
     * @param payload the Object to use as payload
     * @return {@link SendResult}
     */
    public SendResult syncSend(String destination, Object payload) {
        return syncSend(destination, payload, producer.getSendMsgTimeout());
    }

    /**
     * Same to {@link #syncSend(String, Object)} with send timeout specified in addition.
     *
     * @param destination formats: `topicName:tags`
     * @param payload the Object to use as payload
     * @param timeout send timeout with millis
     * @return {@link SendResult}
     */
    public SendResult syncSend(String destination, Object payload, long timeout) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        return syncSend(destination, message, timeout);
    }

    /**
     * Same to {@link #syncSend(String, Message)} with send orderly with hashKey by specified.
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @param hashKey use this key to select queue. for example: orderId, productId ...
     * @return {@link SendResult}
     */
    public SendResult syncSendOrderly(String destination, Message<?> message, String hashKey) {
        return syncSendOrderly(destination, message, hashKey, producer.getSendMsgTimeout());
    }

    /**
     * Same to {@link #syncSendOrderly(String, Message, String)} with send timeout specified in addition.
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @param hashKey use this key to select queue. for example: orderId, productId ...
     * @param timeout send timeout with millis
     * @return {@link SendResult}
     */
    public SendResult syncSendOrderly(String destination, Message<?> message, String hashKey, long timeout) {
        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
            log.error("syncSendOrderly failed. destination:{}, message is null ", destination);
            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
        }
        try {
            long now = System.currentTimeMillis();
            org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
            SendResult sendResult = producer.send(rocketMsg, messageQueueSelector, hashKey, timeout);
            long costTime = System.currentTimeMillis() - now;
            if (log.isDebugEnabled()) {
                log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());
            }
            return sendResult;
        } catch (Exception e) {
            log.error("syncSendOrderly failed. destination:{}, message:{} ", destination, message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * Same to {@link #syncSend(String, Object)} with send orderly with hashKey by specified.
     *
     * @param destination formats: `topicName:tags`
     * @param payload the Object to use as payload
     * @param hashKey use this key to select queue. for example: orderId, productId ...
     * @return {@link SendResult}
     */
    public SendResult syncSendOrderly(String destination, Object payload, String hashKey) {
        return syncSendOrderly(destination, payload, hashKey, producer.getSendMsgTimeout());
    }

    /**
     * Same to {@link #syncSendOrderly(String, Object, String)} with send timeout specified in addition.
     *
     * @param destination formats: `topicName:tags`
     * @param payload the Object to use as payload
     * @param hashKey use this key to select queue. for example: orderId, productId ...
     * @param timeout send timeout with millis
     * @return {@link SendResult}
     */
    public SendResult syncSendOrderly(String destination, Object payload, String hashKey, long timeout) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        return syncSendOrderly(destination, message, hashKey, timeout);
    }

    /**
     * Same to {@link #asyncSend(String, Message, SendCallback)} with send timeout and delay level specified in
     * addition.
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @param sendCallback {@link SendCallback}
     * @param timeout send timeout with millis
     * @param delayLevel level for the delay message
     */
    public void asyncSend(String destination, Message<?> message, SendCallback sendCallback, long timeout,
        int delayLevel) {
        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
            log.error("asyncSend failed. destination:{}, message is null ", destination);
            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
        }
        try {
            org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
            if (delayLevel > 0) {
                rocketMsg.setDelayTimeLevel(delayLevel);
            }
            producer.send(rocketMsg, sendCallback, timeout);
        } catch (Exception e) {
            log.info("asyncSend failed. destination:{}, message:{} ", destination, message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * Same to {@link #asyncSend(String, Message, SendCallback)} with send timeout specified in addition.
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @param sendCallback {@link SendCallback}
     * @param timeout send timeout with millis
     */
    public void asyncSend(String destination, Message<?> message, SendCallback sendCallback, long timeout) {
        asyncSend(destination, message, sendCallback, timeout, 0);
    }

    /**
     * <p> Send message to broker asynchronously. asynchronous transmission is generally used in response time
     * sensitive business scenarios. </p>
     * <p>
     * This method returns immediately. On sending completion, <code>sendCallback</code> will be executed.
     * <p>
     * Similar to {@link #syncSend(String, Object)}, internal implementation would potentially retry up to {@link
     * DefaultMQProducer#getRetryTimesWhenSendAsyncFailed} times before claiming sending failure, which may yield
     * message duplication and application developers are the one to resolve this potential issue.
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @param sendCallback {@link SendCallback}
     */
    public void asyncSend(String destination, Message<?> message, SendCallback sendCallback) {
        asyncSend(destination, message, sendCallback, producer.getSendMsgTimeout());
    }

    /**
     * Same to {@link #asyncSend(String, Object, SendCallback)} with send timeout specified in addition.
     *
     * @param destination formats: `topicName:tags`
     * @param payload the Object to use as payload
     * @param sendCallback {@link SendCallback}
     * @param timeout send timeout with millis
     */
    public void asyncSend(String destination, Object payload, SendCallback sendCallback, long timeout) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        asyncSend(destination, message, sendCallback, timeout);
    }

    /**
     * Same to {@link #asyncSend(String, Message, SendCallback)}.
     *
     * @param destination formats: `topicName:tags`
     * @param payload the Object to use as payload
     * @param sendCallback {@link SendCallback}
     */
    public void asyncSend(String destination, Object payload, SendCallback sendCallback) {
        asyncSend(destination, payload, sendCallback, producer.getSendMsgTimeout());
    }

    /**
     * Same to {@link #asyncSendOrderly(String, Message, String, SendCallback)} with send timeout specified in
     * addition.
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @param hashKey use this key to select queue. for example: orderId, productId ...
     * @param sendCallback {@link SendCallback}
     * @param timeout send timeout with millis
     */
    public void asyncSendOrderly(String destination, Message<?> message, String hashKey, SendCallback sendCallback,
        long timeout) {
        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
            log.error("asyncSendOrderly failed. destination:{}, message is null ", destination);
            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
        }
        try {
            org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
            producer.send(rocketMsg, messageQueueSelector, hashKey, sendCallback, timeout);
        } catch (Exception e) {
            log.error("asyncSendOrderly failed. destination:{}, message:{} ", destination, message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * Same to {@link #asyncSend(String, Message, SendCallback)} with send orderly with hashKey by specified.
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @param hashKey use this key to select queue. for example: orderId, productId ...
     * @param sendCallback {@link SendCallback}
     */
    public void asyncSendOrderly(String destination, Message<?> message, String hashKey, SendCallback sendCallback) {
        asyncSendOrderly(destination, message, hashKey, sendCallback, producer.getSendMsgTimeout());
    }

    /**
     * Same to {@link #asyncSendOrderly(String, Message, String, SendCallback)}.
     *
     * @param destination formats: `topicName:tags`
     * @param payload the Object to use as payload
     * @param hashKey use this key to select queue. for example: orderId, productId ...
     * @param sendCallback {@link SendCallback}
     */
    public void asyncSendOrderly(String destination, Object payload, String hashKey, SendCallback sendCallback) {
        asyncSendOrderly(destination, payload, hashKey, sendCallback, producer.getSendMsgTimeout());
    }

    /**
     * Same to {@link #asyncSendOrderly(String, Object, String, SendCallback)} with send timeout specified in addition.
     *
     * @param destination formats: `topicName:tags`
     * @param payload the Object to use as payload
     * @param hashKey use this key to select queue. for example: orderId, productId ...
     * @param sendCallback {@link SendCallback}
     * @param timeout send timeout with millis
     */
    public void asyncSendOrderly(String destination, Object payload, String hashKey, SendCallback sendCallback,
        long timeout) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        asyncSendOrderly(destination, message, hashKey, sendCallback, timeout);
    }

    /**
     * Similar to <a href="https://en.wikipedia.org/wiki/User_Datagram_Protocol">UDP</a>, this method won't wait for
     * acknowledgement from broker before return. Obviously, it has maximums throughput yet potentials of message loss.
     * <p>
     * One-way transmission is used for cases requiring moderate reliability, such as log collection.
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     */
    public void sendOneWay(String destination, Message<?> message) {
        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
            log.error("sendOneWay failed. destination:{}, message is null ", destination);
            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
        }
        try {
            org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
            producer.sendOneway(rocketMsg);
        } catch (Exception e) {
            log.error("sendOneWay failed. destination:{}, message:{} ", destination, message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * Same to {@link #sendOneWay(String, Message)}
     *
     * @param destination formats: `topicName:tags`
     * @param payload the Object to use as payload
     */
    public void sendOneWay(String destination, Object payload) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        sendOneWay(destination, message);
    }

    /**
     * Same to {@link #sendOneWay(String, Message)} with send orderly with hashKey by specified.
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @param hashKey use this key to select queue. for example: orderId, productId ...
     */
    public void sendOneWayOrderly(String destination, Message<?> message, String hashKey) {
        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
            log.error("sendOneWayOrderly failed. destination:{}, message is null ", destination);
            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
        }
        try {
            org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
            producer.sendOneway(rocketMsg, messageQueueSelector, hashKey);
        } catch (Exception e) {
            log.error("sendOneWayOrderly failed. destination:{}, message:{}", destination, message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * Same to {@link #sendOneWayOrderly(String, Message, String)}
     *
     * @param destination formats: `topicName:tags`
     * @param payload the Object to use as payload
     */
    public void sendOneWayOrderly(String destination, Object payload, String hashKey) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        sendOneWayOrderly(destination, message, hashKey);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        if (producer != null) {
            producer.start();
        }
    }

    @Override
    protected void doSend(String destination, Message<?> message) {
        SendResult sendResult = syncSend(destination, message);
        if (log.isDebugEnabled()) {
            log.debug("send message to `{}` finished. result:{}", destination, sendResult);
        }
    }

    @Override
    protected Message<?> doConvert(Object payload, Map<String, Object> headers, MessagePostProcessor postProcessor) {
        Message<?> message = super.doConvert(payload, headers, postProcessor);
        MessageBuilder<?> builder = MessageBuilder.fromMessage(message);
        builder.setHeaderIfAbsent(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN);
        return builder.build();
    }

    @Override
    public void destroy() {
        if (Objects.nonNull(producer)) {
            producer.shutdown();
        }
    }

    /**
     * Send Spring Message in Transaction
     *
     * @param destination destination formats: `topicName:tags`
     * @param message message {@link org.springframework.messaging.Message}
     * @param arg ext arg
     * @return TransactionSendResult
     * @throws MessagingException
     */
    public TransactionSendResult sendMessageInTransaction(final String destination,
        final Message<?> message, final Object arg) throws MessagingException {
        try {
            if (((TransactionMQProducer) producer).getTransactionListener() == null) {
                throw new IllegalStateException("The rocketMQTemplate does not exist TransactionListener");
            }
            org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
            return producer.sendMessageInTransaction(rocketMsg, arg);
        } catch (MQClientException e) {
            throw RocketMQUtil.convert(e);
        }
    }

    private org.apache.rocketmq.common.message.Message createRocketMqMessage(
        String destination, Message<?> message) {
        Message<?> msg = this.doConvert(message.getPayload(), message.getHeaders(), null);
        return RocketMQUtil.convertToRocketMessage(getMessageConverter(), charset,
            destination, msg);
    }

    private Object doConvertMessage(MessageExt messageExt, Type type) {
        if (Objects.equals(type, MessageExt.class)) {
            return messageExt;
        } else if (Objects.equals(type, byte[].class)) {
            return messageExt.getBody();
        } else {
            String str = new String(messageExt.getBody(), Charset.forName(charset));
            if (Objects.equals(type, String.class)) {
                return str;
            } else {
                // If msgType not string, use objectMapper change it.
                try {
                    if (type instanceof Class) {
                        //if the messageType has not Generic Parameter
                        return this.getMessageConverter().fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) type);
                    } else {
                        //if the messageType has Generic Parameter, then use SmartMessageConverter#fromMessage with third parameter "conversionHint".
                        //we have validate the MessageConverter is SmartMessageConverter in this#getMethodParameter.
                        return ((SmartMessageConverter) this.getMessageConverter()).fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) ((ParameterizedType) type).getRawType(), null);
                    }
                } catch (Exception e) {
                    log.error("convert failed. str:{}, msgType:{}", str, type);
                    throw new RuntimeException("cannot convert message to " + type, e);
                }
            }
        }
    }

    private Type getMessageType(RocketMQLocalRequestCallback rocketMQLocalRequestCallback) {
        Class<?> targetClass = AopProxyUtils.ultimateTargetClass(rocketMQLocalRequestCallback);
        Type matchedGenericInterface = null;
        while (Objects.nonNull(targetClass)) {
            Type[] interfaces = targetClass.getGenericInterfaces();
            if (Objects.nonNull(interfaces)) {
                for (Type type : interfaces) {
                    if (type instanceof ParameterizedType && (Objects.equals(((ParameterizedType) type).getRawType(), RocketMQLocalRequestCallback.class))) {
                        matchedGenericInterface = type;
                        break;
                    }
                }
            }
            targetClass = targetClass.getSuperclass();
        }
        if (Objects.isNull(matchedGenericInterface)) {
            return Object.class;
        }

        Type[] actualTypeArguments = ((ParameterizedType) matchedGenericInterface).getActualTypeArguments();
        if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) {
            return actualTypeArguments[0];
        }
        return Object.class;
    }
}