package com.elab.mq.listener;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.dianping.cat.message.Transaction;
import com.elab.log.utils.CatCrossProcess;
import com.elab.mq.exception.MsgRollBackException;
import com.elab.mq.model.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * 抽象成客户端能够使用的方法
 *
 * @author ： liukx
 * @time ： 2019/4/22 - 18:02
 */
public abstract class AbstractMessageListener implements MessageListener {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired(required = false)
    private ConsumerInterceptor consumerInterceptor = new LoggerConsumerInterceptor();

    private final String MQ_CONSUMER = "MQ_CONSUMER";

    /**
     * 客户端关注的topic消息,实现了该方法,消息监听被触发时会与之匹配。
     *
     * @return
     */
    public abstract String topic();

    /**
     * 客户端关注的topic下的tag子类标签
     *
     * @return
     */
    public String tag() {
        return "*";
    }

    /**
     * 客户端的业务逻辑实现
     *
     * @param message        消息对象
     * @param consumeContext 消息上下文
     * @return
     */
    public abstract Action consume0(MessageModel message, ConsumeContext consumeContext) throws Exception;

    @Override
    public Action consume(Message message, ConsumeContext consumeContext) {
        String topic = message.getTopic();
        String tag = message.getTag();
        String groupName = this.getClass().getSimpleName();

        Transaction t = CatCrossProcess
            .getCrossTransactionMsg(MQ_CONSUMER, topic + "_" + tag + "_" + getClass().getSimpleName(), null);
        MessageModel messageModel = new MessageModel(message);

        Action action = null;
        long start = System.currentTimeMillis();

        try {
            // 前置检查
            boolean isConsumer = this.consumerInterceptor.check(messageModel, groupName);

            //如果已经成功过一次
            if (!isConsumer) {
                t.setSuccessStatus();
                return Action.CommitMessage;
            }

            action = consume0(messageModel, consumeContext);

            long time = System.currentTimeMillis() - start;
            // Manual
            messageModel.setInvokeTime(time);

            t.setSuccessStatus();
            // 可能出现手动提交回滚状态的情况。
            if (action == null || action == Action.ReconsumeLater) {
                // 手动提交或者极端情况返回为null的情况
                this.consumerInterceptor.error(messageModel, groupName, new MsgRollBackException("手动提交回滚"));
            } else {
                this.consumerInterceptor.success(messageModel, groupName);
            }
        } catch (Exception e) {
            t.setStatus(e);
            logger.error("消息处理异常 : ", e);
            messageModel.setInvokeTime(System.currentTimeMillis() - start);
            this.consumerInterceptor.error(messageModel, groupName, e);
            return Action.ReconsumeLater;
        } finally {
            t.complete();
        }
        logger.debug("消息处理结果 : " + action);
        return action;
    }

}
