/**
 * 
 */
package org.springblade.etl.task.init;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

import org.springblade.core.tool.utils.CollectionUtil;
import org.springblade.core.tool.utils.SpringUtil;
import org.springblade.etl.engine.datax.event.JobRunEvent;
import org.springblade.etl.engine.datax.event.JobRunInstance;
import org.springblade.etl.engine.datax.event.JobRunTarget;
import org.springblade.etl.entity.DsyJobInstance;
import org.springblade.etl.entity.JobInstance;
import org.springblade.etl.service.DsyJobInstanceService;
import org.springblade.etl.service.JobInstanceService;
import org.springblade.etl.task.model.DaemonTaskConfig;
import org.springblade.etl.task.model.DaemonTaskType;
import org.springblade.etl.task.thread.DaemonFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

/**
 * @author wookvn
 *
 */
@Slf4j
@Component
@Order(value = 1)
@RequiredArgsConstructor
@ConditionalOnProperty(prefix = "platomix.etl", value = "enabled", havingValue = "true")
public class SystemInit implements ApplicationRunner, DisposableBean {
	
	private final JobInstanceService jobInstanceService;
	
	private final DsyJobInstanceService dsyJobInstanceService;
	
	@Override
	public void run(ApplicationArguments args) throws Exception {
		//作业监控初始化
		List<DaemonTaskConfig> allTaskConfig = new ArrayList<>();
		allTaskConfig.addAll(initEtlTaskMonitor());
		allTaskConfig.addAll(initDsyRunningInstanceMonitor());
		log.info("Number of tasks started: " + allTaskConfig.size());
		DaemonFactory.start(allTaskConfig);
		//待运行的DSY实例放入队列
		initDsyWaitingInstanceMonitor();
	}
	
	@Override
	public void destroy() throws Exception {
		DaemonFactory.shutdown();
	}
	
	/**
	 * 初始化ETL任务实例
	 */
	private List<DaemonTaskConfig> initEtlTaskMonitor() {
		List<JobInstance> jobInstanceList = jobInstanceService.listAllRunningInstanceList();
		List<DaemonTaskConfig> configList = new ArrayList<>(jobInstanceList.size());
		if(CollectionUtil.isNotEmpty(jobInstanceList)) {
			for(JobInstance jobInstance : jobInstanceList) {
				configList.add(new DaemonTaskConfig(jobInstance.getId().toString(), DaemonTaskType.flink_job));
			}
		}
		return configList;
	}
	
	/**
	 * 初始化数据同步任务实例
	 * @return
	 */
	private List<DaemonTaskConfig> initDsyRunningInstanceMonitor() {
		List<DsyJobInstance> jobInstanceList = dsyJobInstanceService.listAllRunningInstanceList();
		List<DaemonTaskConfig> configList = new ArrayList<>(jobInstanceList.size());
		if(CollectionUtil.isNotEmpty(jobInstanceList)) {
			for(DsyJobInstance jobInstance : jobInstanceList) {
				configList.add(new DaemonTaskConfig(jobInstance.getId().toString(), DaemonTaskType.datax_job));
			}
		}
		return configList;
	}
	
	/**
	 * 初始化待执行的数据同步实例
	 */
	private void initDsyWaitingInstanceMonitor() {
		List<DsyJobInstance> jobInstanceList = dsyJobInstanceService.listAllWaitingInstanceList();
		if(CollectionUtil.isNotEmpty(jobInstanceList)) {
			log.info("Number of DSY:Waiting task pushed queue: " + jobInstanceList.size());
			List<JobRunTarget> targets = jobInstanceList.stream().map(item -> new JobRunTarget(item.getId(), item.getInstanceRunType())).collect(Collectors.toList());
			JobRunInstance.getInstance().add(targets);
			SpringUtil.publishEvent(new JobRunEvent());
		}
	}
}
