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.elab.log.utils.CatCrossProcess;
import com.elab.mq.model.MessageModel;
import com.elab.mq.utils.RocketMonitorUtils;
import com.google.common.base.Splitter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * 订阅组件适配器
 *
 * @author ： liukx
 * @time ： 2019/4/22 - 17:59
 */
@Component
public class MessageListenerWrapper implements MessageListener {

    @Autowired(required = false)
    private List<AbstractMessageListener> messageListeners;

    private String defaultRmqGroup = "rmq_group";

    private Logger logger = LoggerFactory.getLogger(MessageListenerWrapper.class);

    @Override
    public Action consume(Message message, ConsumeContext consumeContext) {
        Action result = Action.CommitMessage;
        if (messageListeners != null) {
            try {
                Map<String, String> msgMap = conversionPropertyToMap(message.getUserProperties());
                // 构建上下文消息
                result = CatCrossProcess.buildRemoteMQMsg(defaultRmqGroup, message.getTopic(), msgMap, () -> {
                    MessageModel messageModel = new MessageModel(message);
                    logger.info("消息体:" + message.toString());
                    logger.info("消息内容:" + messageModel.getObject(String.class));
                    long start = System.currentTimeMillis();
                    for (int i = 0; i < messageListeners.size(); i++) {
                        AbstractMessageListener abstractMessageListener = messageListeners.get(i);
                        String currentTopic = message.getTopic();
                        String currentTag = message.getTag();
                        String topic = abstractMessageListener.topic();
                        String tag = abstractMessageListener.tag();
                        if (topic.equals(currentTopic) && ("*".equals(tag) || isContainsTags(tag, currentTag))) {
                            Action actionResult = abstractMessageListener.consume(message, consumeContext);
                            if (actionResult == null || Action.ReconsumeLater == actionResult) {
                                logger.warn(" MQ 消费失败 : " + abstractMessageListener);
                                long time = System.currentTimeMillis() - start;
                                RocketMonitorUtils.sendConsumerMonitorData(message, -1, time);
                                return Action.ReconsumeLater;
                            }
                        }
                    }
                    long time = System.currentTimeMillis() - start;
                    RocketMonitorUtils.sendConsumerMonitorData(message, 1, time);
                    logger.info("消费成功");
                    return Action.CommitMessage;
                });
            } catch (Exception e) {
                logger.error("未知异常", e);
                RocketMonitorUtils.sendConsumerMonitorData(message, -1, -1L);
                return Action.ReconsumeLater;
            }
        }
        return result;
    }

    /**
     * 触发客户端发送的tag是多个的情况
     *
     * @param tags       消费者订阅的tag列表
     * @param currentTag 当前消息携带的tag标签 可能是多个
     * @return
     */
    public static boolean isContainsTags(String tags, String currentTag) {
        // 消费者订阅的tag
        List<String> tagList = Splitter.on("||").trimResults().omitEmptyStrings().splitToList(tags);
        // 消息发送的过来的tag
        List<String> currentTagList = Splitter.on("||").trimResults().omitEmptyStrings().splitToList(currentTag);
        boolean flag = false;
        for (int i = 0; i < currentTagList.size(); i++) {
            String s = currentTagList.get(i);
            if (tagList.contains(s)) {
                flag = true;
                break;
            }
        }
        return flag;
    }

    private Map<String, String> conversionPropertyToMap(Properties userProperties) {
        if (userProperties == null) {
            return null;
        }
        Map<String, String> msgMap = new LinkedHashMap<>();
        userProperties.stringPropertyNames().forEach((key) -> {
            msgMap.put(key, userProperties.getProperty(key));
        });
        return msgMap;
    }

}
