package org.ssssssss.magicapi.rocketmq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ssssssss.magicapi.rocketmq.service.RocketMqService;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author zzy
 */

public class RocketMQConsumer {

    Logger logger = LoggerFactory.getLogger(RocketMQConsumer.class);
    private static Map<String, DefaultMQPushConsumer> consumerMap;

    private final String namesrvAddr;

    private final RocketMqService rocketMqService;

    public RocketMQConsumer(String namesrvAddr, RocketMqService rocketMqService) {
        consumerMap = new HashMap<>();
        this.namesrvAddr = namesrvAddr;
        this.rocketMqService = rocketMqService;
    }

    public void subscribe(String topic, String tag, String scriptName, String script) throws MQClientException {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(topic + "-" + tag);
        defaultMQPushConsumer.setNamesrvAddr(namesrvAddr);
        // 订阅主题和Tag过滤模式，使用*表示全量订阅。
        defaultMQPushConsumer.subscribe(topic, tag);
        defaultMQPushConsumer.setInstanceName(topic + "-" + tag);
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(1);
        // 注册监听器。
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    logger.info("Received message: " + new String(msg.getBody()));
                    if (!rocketMqService.consume(scriptName, script, msg)) {
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        defaultMQPushConsumer.start();
        consumerMap.put(topic + "-" + tag, defaultMQPushConsumer);
    }

    public void unsubscribe(String topic, String tag) {
        DefaultMQPushConsumer defaultMQPushConsumer = consumerMap.get(topic + "-" + tag);
        if (defaultMQPushConsumer != null) {
            defaultMQPushConsumer.unsubscribe(topic);
            defaultMQPushConsumer.shutdown();
            consumerMap.remove(topic + "-" + tag);
        }
    }
}
