package com.kxs.transfer.api.config;

import com.aliyun.dts.subscribe.clients.ConsumerContext;
import com.aliyun.dts.subscribe.clients.DTSConsumer;
import com.aliyun.dts.subscribe.clients.DefaultDTSConsumer;
import com.aliyun.dts.subscribe.clients.common.RecordListener;
import com.kxs.transfer.api.annotation.DtsListener;
import com.kxs.transfer.api.service.KxsCheckpointService;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.scheduling.annotation.Async;

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

/**
 *  DTS 订阅管道配置
 *
 * @author 没秃顶的码农
 * @date 2024-01-19
 */
@Configuration(proxyBeanMethods = false)
@RequiredArgsConstructor
@EnableConfigurationProperties(DTSProperties.class)
public class ConsumerAssignConfig implements ApplicationListener<ContextRefreshedEvent>, DisposableBean {

    private final KxsCheckpointService kxsCheckpointService;

    private final DTSProperties dtsProperties;

    DTSConsumer dtsConsumer = null;

    @Override
    @Async
    public void onApplicationEvent(ContextRefreshedEvent event) {
        System.out.println("async kafka is start...");
        //如果是第二次加载则不处理了
        if(event.getApplicationContext().getParent() != null) {
            return;
        }
        if(dtsConsumer != null){
            return;
        }
        //获取注解类
        Map<String, Object> beanMap;
        beanMap = event.getApplicationContext().getBeansWithAnnotation(DtsListener.class);
        //未扫描到不处理
        if(beanMap.isEmpty()) {
            return;
        }
        Map<String, RecordListener> finalBeanMap =  new HashMap<>();
        beanMap.forEach((name, score) -> finalBeanMap.put(name, (RecordListener)score));
        //初始化消费者
        initDTSClient(finalBeanMap);
    }

    /**
     * 初始化 初始化消费者
     *
     */
    public void initDTSClient(Map<String, RecordListener> beans) {

        ConsumerContext consumerContext = new ConsumerContext(
                dtsProperties.getBrokerUrl(),
                dtsProperties.getKafkaTopic(),
                dtsProperties.getSid(),
                dtsProperties.getUsername(),
                dtsProperties.getPassword(),
                dtsProperties.getInitCheckpoint(),
                ConsumerContext.ConsumerSubscribeMode.ASSIGN);

        //if this parameter is set, force to use the initCheckpoint to initial
        consumerContext.setForceUseCheckpoint(dtsProperties.getIsForceUseInitCheckpoint());

        //add user store
        consumerContext.setUserRegisteredStore(new UserMetaStore(kxsCheckpointService));

        dtsConsumer = new DefaultDTSConsumer(consumerContext);

        dtsConsumer.addRecordListeners(beans);
        dtsConsumer.start();
    }


    @Override
    public void destroy() {
        if(dtsConsumer != null) {
            dtsConsumer.close();
        }
    }
}
