package org.ssssssss.magicapi.rocketmq.service;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.event.EventListener;
import org.ssssssss.magicapi.core.config.MagicConfiguration;
import org.ssssssss.magicapi.core.event.FileEvent;
import org.ssssssss.magicapi.core.event.GroupEvent;
import org.ssssssss.magicapi.core.service.AbstractMagicDynamicRegistry;
import org.ssssssss.magicapi.core.service.MagicResourceStorage;
import org.ssssssss.magicapi.rocketmq.consumer.RocketMQConsumer;
import org.ssssssss.magicapi.rocketmq.model.RocketMqInfo;

/**
 * @author zzy
 */
public class RocketMqMagicDynamicRegistry extends AbstractMagicDynamicRegistry<RocketMqInfo> {
    private final RocketMQConsumer rocketMQConsumer;
    private static final Logger logger = LoggerFactory.getLogger(RocketMqMagicDynamicRegistry.class);

    private final boolean showLog;


    public RocketMqMagicDynamicRegistry(MagicResourceStorage<RocketMqInfo> magicResourceStorage, boolean showLog, RocketMqService rocketMqService,
                                        String namesrvAddr) {
        super(magicResourceStorage);
        this.rocketMQConsumer = new RocketMQConsumer(namesrvAddr, rocketMqService);
        this.showLog = showLog;
    }

    @EventListener(condition = "#event.type == 'rocketmq'")
    public void onFileEvent(FileEvent event) {
        processEvent(event);
    }

    @EventListener(condition = "#event.type == 'rocketmq'")
    public void onGroupEvent(GroupEvent event) {
        processEvent(event);
    }

    @Override
    public boolean register(RocketMqInfo entity) {
        unregister(entity);
        return super.register(entity);
    }

    @Override
    protected boolean register(MappingNode<RocketMqInfo> mappingNode) {
        RocketMqInfo entity = mappingNode.getEntity();
        if (rocketMQConsumer != null) {
            String scriptName = MagicConfiguration.getMagicResourceService().getScriptName(entity);
            try {
                if (entity.isEnabled()) {
                    try {
                        if (showLog) {
                            logger.info("消息队列:[{}]开始执行", scriptName);
                        }
                        rocketMQConsumer.subscribe(entity.getTopic(), entity.getTag(), scriptName, entity.getScript());
                    } catch (Exception e) {
                        logger.error("消息队列执行出错", e);
                    } finally {
                        if (showLog) {
                            logger.info("消息队列:[{}]执行完毕", scriptName);
                        }
                    }
                }
            } catch (Exception e) {
                logger.error("消息队列:[{}]注册失败", scriptName, e);
            }
            logger.debug("注册消息队列:[{},{}]", MagicConfiguration.getMagicResourceService().getScriptName(entity), entity.getTopic());
        }

        return true;
    }

    @Override
    protected void unregister(MappingNode<RocketMqInfo> mappingNode) {
        RocketMqInfo info = mappingNode.getEntity();
        logger.debug("取消注册消息队列:[{}, {}, {}]", info.getName(), info.getPath(), info.getTopic());
        if (rocketMQConsumer != null) {
            try {
                rocketMQConsumer.unsubscribe(info.getTopic(), info.getTag());
            } catch (Exception e) {
                String scriptName = MagicConfiguration.getMagicResourceService().getScriptName(info);
                logger.warn("消息队列:[{}]取消失败", scriptName, e);
            }
        }
    }
}
