package com.elab.mq.rocket.listener;

import com.elab.mq.rocket.exception.MsgProcessException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * tag执行处理规则
 *
 * @author ： liukx
 * @time ： 2019/9/16 - 16:51
 */
public class TagInvokerListener implements InitializingBean {

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

    @Autowired
    private List<TagRocketMQListener> topicList;

    private Map<String, TagRocketMQListener> tagMap;

    private TagRocketMQListener defaultTagRocketMQListener;

    public TagInvokerListener() {
        this.tagMap = new ConcurrentHashMap<>();
    }


    public void process(String tag, String body) throws Exception {
        if (defaultTagRocketMQListener != null) {
            tagMap.getOrDefault(tag, defaultTagRocketMQListener).processMessage(body);
        } else {
            TagRocketMQListener tagRocketMQListener = tagMap.get(tag);
            if (tagRocketMQListener == null) {
                throw new MsgProcessException(" 没有匹配到对应的tags : " + tag);
            }
            tagRocketMQListener.processMessage(body);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        AtomicInteger allIndex = new AtomicInteger();
        topicList.forEach((obj) -> {
            String tag = obj.tag();
            if (tag.equals("*")) {
                defaultTagRocketMQListener = obj;
                allIndex.getAndIncrement();
            } else {
                tagMap.put(tag, obj);
            }
        });

        int count = allIndex.get();
        if (count > 1) {
            logger.warn(" 默认的AbstractRocketMQListener中的tag = * 出现了 {} 个 , 在匹配不上tag的时候 默认匹配 : {}", count, defaultTagRocketMQListener);
        }
    }
}
