/**
 * 
 */
package org.springblade.etl.config;

import java.net.URL;
import java.util.Objects;

import org.springblade.etl.engine.flink.client.rest.FlinkRestClient;
import org.springblade.etl.properties.EtlProperties;
import org.springframework.beans.factory.config.ConstructorArgumentValues;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.env.Environment;
import org.springframework.core.type.AnnotationMetadata;

import lombok.extern.slf4j.Slf4j;

/**
 * @author wookvn
 * 根据配置文件动态初始化 FlinkRestClient
 */
@Slf4j
@Configuration
@EnableConfigurationProperties(EtlProperties.class)
@Import(EtlImportConfig.EtlFlinkClientImportConfig.class)
public class EtlImportConfig {
	
	public static class EtlFlinkClientImportConfig  implements ImportBeanDefinitionRegistrar, EnvironmentAware {
		private EtlProperties etlProperties;
		
		@Override
		public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry beanDefinitionRegistry) {
			for(String webInterfaceUrl : etlProperties.getFlinkWebInterfaceUrls()) {
				try {
					URL url = new URL(webInterfaceUrl);
					//注册Bean
					RootBeanDefinition beanDefinition = new RootBeanDefinition();
					beanDefinition.setBeanClass(FlinkRestClient.class);
					ConstructorArgumentValues constructorArgumentValues = new ConstructorArgumentValues();
					constructorArgumentValues.addIndexedArgumentValue(0, url.getHost());
					constructorArgumentValues.addIndexedArgumentValue(1, url.getPort());
					constructorArgumentValues.addIndexedArgumentValue(2, new org.apache.flink.configuration.Configuration());
		            beanDefinition.setConstructorArgumentValues(constructorArgumentValues);
		            String beanName = EtlImportConfig.flinkClientBeanName(url);
		            log.info("初始化FlinkRestClient|BeanName={}", beanName);
		            beanDefinitionRegistry.registerBeanDefinition(beanName, beanDefinition);
				} catch (Exception e) {
					throw new UnsupportedOperationException("初始化FlinkRestClient失败", e);
				}
			}
		}
		
		@Override
		public void setEnvironment(Environment environment) {
			String propertiesPrefix = Objects.requireNonNull(AnnotationUtils.getAnnotation(EtlProperties.class, ConfigurationProperties.class)).prefix();
			etlProperties = Binder.get(environment).bind(propertiesPrefix, EtlProperties.class).get();
		}
	}
	
	/**
	 * 
	 * @param url
	 * @return
	 */
	public static String flinkClientBeanName(URL url) {
		return String.format("restClient_%s_%s", url.getHost(), url.getPort()).replaceAll("\\.", "");
	}
}
