package com.xxl.job.admin.core.jobbean;

import com.xxl.job.admin.core.model.XxlJobGroup;
import com.xxl.job.admin.core.model.XxlJobInfo;
import com.xxl.job.admin.core.model.XxlJobLog;
import com.xxl.job.admin.core.thread.JobMonitorHelper;
import com.xxl.job.admin.core.thread.JobRegistryHelper;
import com.xxl.job.admin.core.schedule.DynamicSchedulerUtil;
import com.xxl.job.core.biz.ExecutorBiz;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
import com.xxl.job.core.registry.RegistHelper;
import com.xxl.job.core.rpc.netcom.NetComClientProxy;
import org.apache.commons.lang.StringUtils;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.JobKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.quartz.QuartzJobBean;

import java.text.MessageFormat;
import java.util.*;

/**
 * http job bean
 * “@DisallowConcurrentExecution” diable concurrent, thread size can not be only one, better given more
 * @author xuxueli 2015-12-17 18:20:34
 */
//@DisallowConcurrentExecution
public class RemoteHttpJobBean extends QuartzJobBean {
	private static Logger logger = LoggerFactory.getLogger(RemoteHttpJobBean.class);

	@Override
	protected void executeInternal(JobExecutionContext context)
			throws JobExecutionException {
		JobKey jobKey = context.getTrigger().getJobKey();
		
		XxlJobInfo jobInfo = DynamicSchedulerUtil.xxlJobInfoDao.load(Integer.valueOf(jobKey.getGroup()), jobKey.getName());
		// save log
		XxlJobLog jobLog = new XxlJobLog();
		jobLog.setJobGroup(jobInfo.getJobGroup());
		jobLog.setJobName(jobInfo.getJobName());
		DynamicSchedulerUtil.xxlJobLogDao.save(jobLog);
		logger.debug(">>>>>>>>>>> xxl-job trigger start, jobId:{}", jobLog.getId());

        // admin address
        List<String> adminAddressList = JobRegistryHelper.discover(RegistHelper.RegistType.ADMIN.name(), RegistHelper.RegistType.ADMIN.name());
		Set<String> adminAddressSet = new HashSet<String>();
        if (adminAddressList!=null) {
            adminAddressSet.addAll(adminAddressList);
        }
        adminAddressSet.add(DynamicSchedulerUtil.getCallbackAddress());

		// update trigger info 1/2
		jobLog.setTriggerTime(new Date());

		// trigger request
		TriggerParam triggerParam = new TriggerParam();
		triggerParam.setJobGroup(String.valueOf(jobInfo.getJobGroup()));
		triggerParam.setJobName(jobInfo.getJobName());
		triggerParam.setExecutorHandler(jobInfo.getExecutorHandler());
		triggerParam.setExecutorParams(jobInfo.getExecutorParam());
		triggerParam.setGlueSwitch((jobInfo.getGlueSwitch()==0)?false:true);
		triggerParam.setLogAddress(adminAddressSet);
		triggerParam.setLogId(jobLog.getId());
		triggerParam.setLogDateTim(jobLog.getTriggerTime().getTime());

		// parse address
		List<String> addressList = new ArrayList<String>();
		XxlJobGroup group = DynamicSchedulerUtil.xxlJobGroupDao.load(Integer.valueOf(jobInfo.getJobGroup()));
		if (group!=null) {
			addressList = JobRegistryHelper.discover(RegistHelper.RegistType.EXECUTOR.name(), group.getAppName());
		}

		// failover trigger
		ReturnT<String> responseModel = failoverTrigger(addressList, triggerParam, jobLog);
		jobLog.setExecutorHandler(jobInfo.getExecutorHandler());
		jobLog.setExecutorParam(jobInfo.getExecutorParam());
		logger.info(">>>>>>>>>>> xxl-job failoverTrigger response, jobId:{}, responseModel:{}", jobLog.getId(), responseModel.toString());
		
		// update trigger info 2/2
		jobLog.setTriggerStatus(responseModel.getCode()+"");
		jobLog.setTriggerMsg(responseModel.getMsg());
		DynamicSchedulerUtil.xxlJobLogDao.updateTriggerInfo(jobLog);

		// monitor triger
		JobMonitorHelper.monitor(jobLog.getId());
		
		logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId());
    }
	
	
	/**
	 * failover for trigger remote address
	 * @return
	 */
	public ReturnT<String> failoverTrigger(List<String> addressList, TriggerParam triggerParam, XxlJobLog jobLog){
		 if (addressList==null || addressList.size() < 1) {
			 return new ReturnT<String>(ReturnT.FAIL_CODE, "Trigger error, <br>>>>[address] is null <br><hr>");
		} else if (addressList.size() == 1) {
			 String address = addressList.get(0);
			 // store real address
			 jobLog.setExecutorAddress(address);

			 // real trigger
			 ExecutorBiz executorBiz = null;
			 try {
				 executorBiz = (ExecutorBiz) new NetComClientProxy(ExecutorBiz.class, address).getObject();
			 } catch (Exception e) {
				 e.printStackTrace();
				 return new ReturnT<String>(ReturnT.FAIL_CODE, e.getMessage());
			 }
			 ReturnT<String> runResult = executorBiz.run(triggerParam);

			 String failoverMessage = MessageFormat.format("Trigger running, <br>>>>[address] : {0}, <br>>>>[code] : {1}, <br>>>>[msg] : {2} <br><hr>",
					 address, runResult.getCode(), runResult.getMsg());
			 runResult.setMsg(runResult.getMsg() + failoverMessage);
			 return runResult;
		 } else {
			
			// for ha
			Collections.shuffle(addressList);

			// for failover
			String failoverMessage = "";
			for (String address : addressList) {
				if (StringUtils.isNotBlank(address)) {


                    ExecutorBiz executorBiz = null;
                    try {
                        executorBiz = (ExecutorBiz) new NetComClientProxy(ExecutorBiz.class, address).getObject();
                    } catch (Exception e) {
                        e.printStackTrace();
                        return new ReturnT<String>(ReturnT.FAIL_CODE, e.getMessage());
                    }

                    // beat check
					ReturnT<String> beatResult = executorBiz.beat();
					failoverMessage += MessageFormat.format("BEAT running, <br>>>>[address] : {0}, <br>>>>[code] : {1}, <br>>>>[msg] : {2} <br><hr>",
							address, beatResult.getCode(), beatResult.getMsg());

					// beat success, trigger do
					if (beatResult.getCode() == ReturnT.SUCCESS_CODE) {
						// store real address
						jobLog.setExecutorAddress(address);

						// real trigger
						ReturnT<String> runResult = executorBiz.run(triggerParam);

						failoverMessage += MessageFormat.format("Trigger running, <br>>>>[address] : {0}, <br>>>>[status] : {1}, <br>>>>[msg] : {2} <br><hr>",
								address, runResult.getCode(), runResult.getMsg());
						runResult.setMsg( runResult.getMsg() + failoverMessage);
						return runResult;
					}

				}
			}

			return new ReturnT<String>(ReturnT.FAIL_CODE, failoverMessage);
		}
	}

	
}