package com.duiba.tuia.youtui.web;

import ch.qos.logback.classic.LoggerContext;
import cn.com.duibaboot.ext.autoconfigure.logger.logback.appender.KafkaAppender;
import cn.com.duibaboot.ext.autoconfigure.logger.logback.appender.KafkaAppenderProperties;
import com.google.common.base.Strings;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.LoggingProducerListener;

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

/**
 * Created by hujinliang on 2018/2/25.
 */
@Configuration
@EnableConfigurationProperties(KafkaAppenderProperties.class)
public  class ExtraKafkaAppenderConfiguration{

    @Autowired
    private KafkaAppenderProperties kafkaAppenderProperties;

    private KafkaTemplate extraKafkaTemplate;

    private boolean flag = true;

    @Value("${spring.kafka.bootstrap-servers-extra}")
    private String extraKafkaServer;

    private Logger logger = LoggerFactory.getLogger(getClass());
    @EventListener(ContextRefreshedEvent.class)
    @SuppressWarnings("squid:S3776")
    public void onApplicationEvent(ContextRefreshedEvent applicationStartedEvent){
        if(flag){
            if(extraKafkaTemplate == null&&!Strings.isNullOrEmpty(extraKafkaServer)){
                Map<String, Object> props = new HashMap<>();
                props.put("bootstrap.servers", extraKafkaServer);//TODO 填入kafka地址
                props.put("request.timeout.ms", "30000");
                props.put("metadata.fetch.timeout.ms", "5000");
                props.put("key.serializer",StringSerializer.class);
                props.put("value.serializer",StringSerializer.class);
                extraKafkaTemplate = new KafkaTemplate(new DefaultKafkaProducerFactory<>(props));
                extraKafkaTemplate.setProducerListener(new LoggingProducerListener<>());
            }
            try {
                if(kafkaAppenderProperties.getPatterns() == null || kafkaAppenderProperties.getPatterns().length == 0){
                    return;
                }

                LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
                Multimap<String, KafkaAppenderProperties.Pattern> logger2PatternsMap = ArrayListMultimap.create();
                for(KafkaAppenderProperties.Pattern pattern : kafkaAppenderProperties.getPatterns()){
                    if(StringUtils.isBlank(pattern.getLoggerName())){
                        throw new IllegalStateException("config `duiba.logback.appender.kafka[*].logger-name` must not be blank");
                    }
                    if(StringUtils.isBlank(pattern.getTopic())){
                        throw new IllegalStateException("config `duiba.logback.appender.kafka[*].topic` must not be blank");
                    }
                    logger2PatternsMap.put(pattern.getLoggerName(), pattern);
                }

                for(String loggerName : logger2PatternsMap.keySet()){
                    Collection<KafkaAppenderProperties.Pattern> patterns = logger2PatternsMap.get(loggerName);
                    if(CollectionUtils.isEmpty(patterns)){
                        continue;
                    }
                    ch.qos.logback.classic.Logger loggerInstantiation = loggerContext.getLogger(loggerName);
                    KafkaAppender appender = new KafkaAppender<>(extraKafkaTemplate, patterns);
                    appender.setContext(loggerContext);
                    appender.start();
                    loggerInstantiation.addAppender(appender);
                }

            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }finally {
                flag = false;
            }
        }
    }

}
