package com.ms.biz.quque;

import com.jinritemai.cloud.base.core.util.LogUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;
import java.util.Objects;

@Slf4j
@Component
public class MqListener implements ApplicationListener<ContextRefreshedEvent> {
    @Resource
    private ApplicationContext applicationContext;

    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        LogUtils.setLogId("MqListener_" + System.currentTimeMillis());
        Map<String, Object> beansWithAnnotationMap = this.applicationContext.getBeansWithAnnotation(ConsumerListener.class);
        log.info("beansWithAnnotationMap: " + beansWithAnnotationMap);
        for (Map.Entry<String, Object> entry: beansWithAnnotationMap.entrySet()) {
            MessageListenerConcurrently concurrently = (MessageListenerConcurrently)entry.getValue();
            ConsumerListener annotation = concurrently.getClass().getAnnotation(ConsumerListener.class);
            if (Objects.equals(annotation.consumerGroup(), "")) {
                log.error("Consumer " + entry.getKey() + "'s consumerGroup is empty, skip!");
                continue;
            }
            if (Objects.equals(annotation.topic(), "")) {
                log.error("Consumer " + entry.getKey() + "'s topic is empty, skip!");
                continue;
            }

            String topic = Topic.rebuild(annotation.topic());
            String group = annotation.consumerGroup();

            log.info("Start consumer: " + entry.getKey() + " group:" + group + " topic:" + topic);
            DefaultMQPushConsumer consumer = MqFactory.newMqConsumer(
                    group,
                    topic,
                    (MessageListenerConcurrently)entry.getValue(),
                    defaultMQPushConsumer -> {
                        if (annotation.consumeThreadMax() >= 0) {
                            defaultMQPushConsumer.setConsumeThreadMax(annotation.consumeThreadMax());
                        }
                        if (annotation.consumeThreadMin() >= 0) {
                            defaultMQPushConsumer.setConsumeThreadMin(annotation.consumeThreadMin());
                        }
                    }
            );
            log.info("consumerInfo: " + consumer);;
        }
    }
}
