/*
 * 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 org.apache.rocketmq.client.Validators;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
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.MessageBatch;
import org.apache.rocketmq.common.message.MessageClientIDSetter;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.support.DelayMode;
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;

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.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutorService;

@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 DefaultLitePullConsumer consumer;

    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 DefaultLitePullConsumer getConsumer() {
        return consumer;
    }

    public void setConsumer(DefaultLitePullConsumer consumer) {
        this.consumer = consumer;
    }

    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;
    }

    public void setAsyncSenderExecutor(ExecutorService asyncSenderExecutor) {
        this.producer.setAsyncSenderExecutor(asyncSenderExecutor);
    }

    /**
     * @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 be 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 be 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 be 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 be 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 be 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 be 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 be 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 be 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 be 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 be 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 be 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 be 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
     *
     * @param destination formats: `topicName:tags`
     * @param messages Collection of {@link org.springframework.messaging.Message}
     * @return {@link SendResult}
     */
    public <T extends Message> SendResult syncSend(String destination, Collection<T> messages) {
        return syncSend(destination, messages, producer.getSendMsgTimeout());
    }

    /**
     * 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 delay time specified in addition.
     * This function is only valid when the broker version is 5.0 or above
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @param delayTime delay time in seconds for message
     * @return {@link SendResult}
     */
    public SendResult syncSendDelayTimeSeconds(String destination, Message<?> message, long delayTime) {
        return syncSend(destination, message, producer.getSendMsgTimeout(), delayTime, DelayMode.DELAY_SECONDS);
    }

    /**
     * Same to {@link #syncSend(String, Object)} with send delayTime specified in addition.
     * This function is only valid when the broker version is 5.0 or above
     *
     * @param destination formats: `topicName:tags`
     * @param payload the Object to use as payload
     * @param delayTime delay time in seconds for message
     * @return {@link SendResult}
     */
    public SendResult syncSendDelayTimeSeconds(String destination, Object payload, long delayTime) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        return syncSend(destination, message, producer.getSendMsgTimeout(), delayTime, DelayMode.DELAY_SECONDS);
    }

    /**
     * Same to {@link #syncSend(String, Message)} with send delay time specified in addition.
     * This function is only valid when the broker version is 5.0 or above
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @param delayTime delay time in millisecond for message
     * @return {@link SendResult}
     */
    public SendResult syncSendDelayTimeMills(String destination, Message<?> message, long delayTime) {
        return syncSend(destination, message, producer.getSendMsgTimeout(), delayTime, DelayMode.DELAY_MILLISECONDS);
    }

    /**
     * Same to {@link #syncSend(String, Object)} with send delayTime specified in addition.
     * This function is only valid when the broker version is 5.0 or above
     *
     * @param destination formats: `topicName:tags`
     * @param payload the Object to use as payload
     * @param delayTime delay time in millisecond for message
     * @return {@link SendResult}
     */
    public SendResult syncSendDelayTimeMills(String destination, Object payload, long delayTime) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        return syncSend(destination, message, producer.getSendMsgTimeout(), delayTime, DelayMode.DELAY_MILLISECONDS);
    }


    /**
     * Same to {@link #syncSend(String, Message)} with send in a delivered time.
     * This function is only valid when the broker version is 5.0 or above
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @param deliverTimeMills delivered time in millisecond for message
     * @return {@link SendResult}
     */
    public SendResult syncSendDeliverTimeMills(String destination, Message<?> message, long deliverTimeMills) {
        return syncSend(destination, message, producer.getSendMsgTimeout(), deliverTimeMills, DelayMode.DELIVER_TIME_MILLISECONDS);
    }

    /**
     * Same to {@link #syncSend(String, Object)} with send in a delivered time.
     * This function is only valid when the broker version is 5.0 or above
     *
     * @param destination formats: `topicName:tags`
     * @param payload the Object to use as payload
     * @param deliverTimeMills delivered time in millisecond for message
     * @return {@link SendResult}
     */
    public SendResult syncSendDeliverTimeMills(String destination, Object payload, long deliverTimeMills) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        return syncSend(destination, message, producer.getSendMsgTimeout(), deliverTimeMills, DelayMode.DELIVER_TIME_MILLISECONDS);
    }

    /**
     * Same to {@link #syncSend(String, Message)} with send timeout and delay time specified in addition.
     * This function is only valid when the broker version is 5.0 or above
     *
     * @param destination formats: `topicName:tags`
     * @param message {@link org.springframework.messaging.Message}
     * @param timeout send timeout with millis
     * @param delayTime delay time for message
     * @return {@link SendResult}
     */
    private SendResult syncSend(String destination, Message<?> message, long timeout, long delayTime, DelayMode mode) {
        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");
        }
        if (Objects.isNull(mode)) {
            throw new IllegalArgumentException("`delay mode` can be null");
        }
        if (delayTime <= 0) {
            throw new IllegalArgumentException("`delayTime` should be greater than 0");
        }
        try {
            long now = System.currentTimeMillis();
            org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
            switch (mode) {
                case DELAY_SECONDS:
                    rocketMsg.setDelayTimeSec(delayTime);
                    break;
                case DELAY_MILLISECONDS:
                    rocketMsg.setDelayTimeMs(delayTime);
                    break;
                case DELIVER_TIME_MILLISECONDS:
                    rocketMsg.setDeliverTimeMs(delayTime);
                    break;
                default:
                    log.warn("delay mode: {} not support", mode);
            }
            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:{}, detail exception info: ", destination, message, e);
            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:{}, detail exception info: ", destination, message, e);
            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.
     *
