/**
 * 
 */
package org.springblade.etl.engine.datax.event;

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

import org.springblade.core.tool.utils.CollectionUtil;
import org.springblade.etl.dto.dsy.RunningInstanceServerDto;
import org.springblade.etl.properties.DsyProperties;
import org.springblade.etl.properties.DsyProperties.ServerInfo;
import org.springblade.etl.service.DsyJobInstanceService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
import org.springframework.core.annotation.Order;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

/**
 * @author wookvn
 *
 */
@Component
public class JobRunListener {
	
	@Autowired
	private DsyProperties dsyConfig;
	
	@Autowired
	private DsyJobInstanceService dsyJobInstanceService;
	
	@Async
	@Order
	@EventListener(JobRunEvent.class)
	public void jobRun(JobRunEvent jobRunEvent) {
		List<RunningInstanceServerDto> instanceServerList = dsyJobInstanceService.countServerByRunningInstance();
		List<ServerInfo> sshServers = dsyConfig.getSshServers();
		List<JobRunTarget> jobRunTargetList = _cal(instanceServerList, sshServers);
		if(jobRunTargetList.isEmpty()) {
			return;
		}
		for(JobRunTarget jobRunTarget : jobRunTargetList) {
			dsyJobInstanceService.runJobInstance(jobRunTarget.getJobInstanceId(), jobRunTarget.getRunType(), jobRunTarget.getServerInfo());
		}
	}
	
	/**
	 * 
	 * @param instanceServerList
	 * @param sshServers
	 * @return
	 */
	private List<JobRunTarget> _cal(List<RunningInstanceServerDto> instanceServerList, List<ServerInfo> sshServers) {
		Map<String, Long> serverCntMap = CollectionUtil.isNotEmpty(instanceServerList) 
				? instanceServerList.stream().collect(Collectors.toMap(RunningInstanceServerDto::getServerIp, RunningInstanceServerDto::getCnt))
				: new HashMap<>();
		List<JobRunTarget> jobRunTargetList = new ArrayList<>();
		symbol: for(ServerInfo serverInfo : sshServers) {
			Long usedNum = serverCntMap.getOrDefault(serverInfo.getIp(), 0L);
			int canRunNum = serverInfo.getMaxTaskNum() - usedNum.intValue();
			if(canRunNum > 0) {
				for(int i = 0; i < canRunNum; i++) {
					JobRunTarget jobRunTarget = JobRunInstance.getInstance().get();
					if(jobRunTarget == null) {
						break symbol;
					}
					jobRunTarget.setServerInfo(serverInfo);
					jobRunTargetList.add(jobRunTarget);
				}
			}
		}
		return jobRunTargetList;
	}
}
