path: root/src/com/nms/server/thread/mission/MissionPollingThread.java

package com.nms.server.thread.mission;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;

import com.nms.server.bean.LoopMissionRoundInfo;
import com.nms.server.bean.MissionResult;
import com.nms.server.bean.MissionStateTable;
import com.nms.server.bean.Task1;
import com.nms.server.bean.Task4;
import com.nms.server.bean.Task6;
import com.nms.server.common.Common;
import com.nms.server.common.Constants;
import com.nms.server.dao.CommonDao;
import com.nms.server.service.EmailService;
import com.nms.server.service.UpgradeService;
import com.nms.server.util.ExceptionPrintUtils;
import com.socket.utils.FileComment;

public class MissionPollingThread implements Runnable{
	private final Logger logger = Logger.getLogger(MissionPollingThread.class);
	private List<MissionStateTable> missionList = null;
	private List<MissionStateTable> missionList2 = new ArrayList<MissionStateTable>();
	private String threadName;
	private Integer testNum = 1;

	/**
	 * 线程单例
	 * @time Nov 15, 2011-7:15:34 PM
	 * @return
	 */
	public MissionPollingThread (String threadName,final Integer testNum,final List<MissionStateTable> missionList) {
		this.missionList = missionList; 
		this.testNum = testNum;
		this.threadName = threadName;
	}
	public void run() {
		Thread.currentThread().setName(threadName);
		
		//-- 结束无意义操作
		if(testNum==null || missionList == null || missionList.size()==0){
			logger.warn("The task queue is empty and does not need to be executed");
			return;
		}
		
		//-- 循环下发任务
		try {
			while(missionList.size()>0){
				MissionStateTable mission = missionList.get(0);
				Thread.currentThread().setName(threadName+" Id:>"+mission.getMissionId());
				switch (mission.getMissionType().intValue()) {
				case 1:
					forTask1(mission);
					break;
					
				case 4:
					if(mission.getLoopFlag()!=null && mission.getLoopFlag().longValue()==1l){	//周期任务
						forLoopMission(mission);
					}else{	//非周期任务
						forTask4(mission);
					}
					break;
					
				case 6:
					forTask6(mission);
					break;
				
				default:
					break;
				}
				missionList2.add(mission);
				missionList.remove(0);
			}
			
			//-- 创建下一次任务轮询
			if(testNum.intValue() < Constants.MISSION_RELEASE_TIMES.intValue()){
				logger.info("开始第“"+(testNum+1)+"”次任务轮询");
				Common.scheduled.schedule(new MissionPollingThread("第“"+(testNum+1)+"”次轮询线程",(testNum+1),missionList2), Constants.MISSION_RELEASE_PERIOD, TimeUnit.SECONDS);
				missionList2 = new ArrayList<MissionStateTable>();
			}else{
				missionList2.clear();
			}
		}catch (Exception e) {
			logger.error(ExceptionPrintUtils.printExceptionStack(e));
		}
	}
	
	/**
	 * 文件推送任务执行
	 * @time Nov 11, 2011-6:55:33 PM
	 * @param mission
	 */
	public void forTask1(MissionStateTable mission) {
		CommonDao dao = null;			//数据库链接
		try {
			dao = new CommonDao();
			UpgradeService service = new UpgradeService(dao);	//专属Service
			
			//-- 任务结果查询和推送文件
			Task1 task1 = service.getTask1ById(mission.getMissionId());		//查询任务具体信息
			List<FileComment> fileCommentsList = service.getMissionFileInformations(mission.getMissionId());
			ArrayList<MissionResult> mrList = service.getMissionResultsForTask(mission.getMissionId(),mission.getMissionType().intValue());
			
			//-- 遍历节点结果集合
			for (MissionResult result : mrList) {
				
				//-- 针对节点  进行文件推送
				Common.sendMissionInfo(mission,new NMSClientTask1(result.getNodeIp(),Constants.SSL_CLIENT_PORT,mission,task1,fileCommentsList),result.getUuid(), testNum,null,null);
			}
		} catch (Exception e) {
			logger.error("Task Id:>"+mission.getMissionId()+" task execution failure", e);
		}finally{
			if(dao!=null){
				dao.close();
				dao=null;
			}
		}
	}
	
//public static void main(String [] args) {
//	List<MissionStateTable> missionList = new LinkedList<MissionStateTable>();
//	CommonDao dao = new CommonDao();			//数据库链接
//	UpgradeService service = new UpgradeService(dao);	//专属Service
//	missionList.add(service.getMessionById(501l));
//	MissionPollingThread thread = new MissionPollingThread ("1",1,missionList);
//	thread.run();
//}
	/**
	 * 命令执行任务执行
	 * @time Nov 11, 2011-6:55:33 PM
	 * @param mission
	 */
	public void forTask4(MissionStateTable mission) {
		CommonDao dao = null;			//数据库链接
		try {
			dao = new CommonDao();
			UpgradeService service = new UpgradeService(dao);	//专属Service
			
			//-- 任务结果查询和推送文件
			Task4 task4 = service.getTask4ById(mission.getMissionId());		//查询任务具体信息
			ArrayList<MissionResult> mrList = service.getMissionResultsForTask(mission.getMissionId(),mission.getMissionType().intValue());
			
			//-- 遍历节点结果集合
			for (MissionResult result : mrList) {
				//-- 针对节点  进行文件推送
				Common.sendMissionInfo(mission,new NMSClientTask4(result.getNodeIp(),Constants.SSL_CLIENT_PORT,mission,task4,result),result.getUuid(), testNum,null,null);
			}
		} catch (Exception e) {
			logger.error("Task Id:>"+mission.getMissionId()+" task execution failure", e);
		}finally{
			if(dao!=null){
				dao.close();
				dao=null;
			}
		}
	}
	
	/**周期任务代码*/
	/**
	 * 业务梳理：
	 * 周期任务操作 ：执行操作状态为0、1 ，撤消操作状态为5，（完成状态时3 此项无需考虑）
	 * 执行状态时：
	 * 	检查并初始化周期任务的轮次（有效轮次 即 轮次起始时间 大于 当前系统时间 的轮次信息条数 <=10）
	 * 	对 当前系统时间 所在的轮次的任务结果初始化
	 * 	对 当前系统时间 所在的轮次的任务结果指定的结果节点进行下发
	 * 撤销状态时：
	 *  对 当前系统时间 所在的轮次的任务结果指定的节点，通知取消周期任务
	 * */
	private void forLoopMission(MissionStateTable mission){
		/**方法执行 参数校验*/
		if(mission == null || mission.getMissionState() == null){
			return ;
		}
		
		CommonDao dao = null;
		SimpleDateFormat format = new SimpleDateFormat(Constants.COMMON_DATE_FORMAT);
		
		try {
			dao = new CommonDao();
			UpgradeService service = new UpgradeService(dao);
			EmailService emailService = new EmailService(dao);
			//- 任务参数的有效性校验
			Task4 task4 = service.getTask4ById(mission.getMissionId());		//查询任务具体信息
			if(task4 == null || StringUtils.isEmpty(task4.getCommandName())|| task4.getCommandType() == 0 ){
//				service.updateMissionState(mission.getMissionId(), 4,format.format(System.currentTimeMillis())+" 任务参数为空，任务无法执行");
//				emailService.sendEmailForMission(mission.getMissionId(),"任务执行失败"," 任务参数为空，任务无法执行");
//				service.updateMissionState(mission.getMissionId(), 4,format.format(System.currentTimeMillis())+" The task parameter is empty and the task cannot be executed");
				service.updateMissionState(mission.getMissionId(), 4,format.format(System.currentTimeMillis())+" i18n_server.MissionPollingThread.noParam_n81i");
//				emailService.sendEmailForMission(mission.getMissionId(),"Task execution failure","The task parameter is empty and the task cannot be executed");
				emailService.sendEmailForMission(mission.getMissionId(),"i18n_server.MissionPollingThread.missionFail_n81i","i18n_server.MissionPollingThread.noParam_n81i");
				logger.warn("The task parameter is empty and the task cannot be executed");
				/***告警功能****/
				return ;
			}
			
			//- 任务时间的有效性校验
			if(mission.getEndTime()==null || mission.getEndTime().getTime() < System.currentTimeMillis() ){
				if(mission.getMissionState().longValue() ==2l){
//					service.updateMissionState(mission.getMissionId(), 3,format.format(System.currentTimeMillis())+" 已超过任务结束时间 任务已完成");
//					service.updateMissionState(mission.getMissionId(), 3,format.format(System.currentTimeMillis())+" The task has exceeded the end time ,and the task execution has ended");
					service.updateMissionState(mission.getMissionId(), 3,format.format(System.currentTimeMillis())+"i18n_server.MissionPollingThread.outEndTime_n81i");
//					emailService.sendEmailForMission(mission.getMissionId(),"任务执行失败"," 已超过任务结束时间 任务已完成");
					emailService.sendEmailForMission(mission.getMissionId(),"i18n_server.MissionPollingThread.missionFail_n81i","i18n_server.MissionPollingThread.outEndTime_n81i");
					logger.warn("The task has exceeded the end time ,and the task execution has ended");
				}
				/***告警功能****/
				return ;
			}
			//- 任务下发 准备操作
			LoopMissionRoundInfo roundInfo = null;
			ArrayList<MissionResult> mrList = null;
			
			//- 任务状态判断 已创建（1）  执行中（2） 准备撤销（5） 撤销中（6）
			logger.debug(mission.getMissionId()+" 周期任务状态"+"  "+mission.getMissionState());
			
			mission.setNodeGroupsId(task4.getNodeGroupsId());	//保存节点组信息 
			mission.setNodeIpsId(task4.getNodeIpsId());		//保存节点信息 
			
			//-- 已创建操作（1）
			if (mission.getMissionState().longValue() == 1l){	//新任务（已创建）的任务下发操作
				
				service.checkAndInitMissionCycle(mission);// 轮次校验和初始化
				List<LoopMissionRoundInfo> roundInfoList = Common.getLoopMissionRoundInfoList().get(mission.getMissionId());
				
				//-- 获取当前轮次信息并初始化当前轮次的任务结果信息
				if(roundInfoList !=null && roundInfoList.size()>0){
					
					//-- 获取当前周期轮次
					for (Iterator<LoopMissionRoundInfo> roundIte = roundInfoList.iterator(); roundIte.hasNext();) {
						roundInfo = roundIte.next();
						if(roundInfo.getEndTime().getTime() > System.currentTimeMillis()){
							break;
						}
					}

					//-- 预置状态的轮次，初始化当前轮次的结果数据
					if(roundInfo!=null && roundInfo.getMissionState()!=null && roundInfo.getMissionState().longValue()==0l){	//0 预置状态，未初始化当前轮次的任务结果信息
						
						service.initLoopMissionResults2(roundInfo.getCurMissionId(), task4.getNodeGroupsId(),task4.getNodeIpsId(),3);//初始化可下发的结果信息 3 未下发
//						service.updateLoopMissionState(roundInfo.getCurMissionId(), 2);	//更新 任务轮次信息状态
						roundInfo.setMissionState(2l);	//更新缓存中的任务轮次信息状态
						mrList = service.getMissionResultsForTask(roundInfo.getCurMissionId(),mission.getMissionType().intValue());
					}else{
						
					}
				}
				
				//任务状态变更
//				if(mrList== null || mrList.size()==0){	//未初始化到当前轮次的结果信息
//					
//					//- 变更任务状态 1（以创建） -> 4（未能执行）
//					service.updateMissionState(mission.getMissionId(), 4,format.format(System.currentTimeMillis())+" 未初始化到可下发节点的结果信息，任务无法执行");
//					mission.setMissionState(4l);
//					return;
//				}else{	

//				}
				
				//- 变更任务状态 1（以创建） -> 2（正在执行）
//				service.updateMissionState(mission.getMissionId(), 2,format.format(System.currentTimeMillis())+" 任务执行开始");
				service.updateMissionState(mission.getMissionId(), 2,format.format(System.currentTimeMillis())+" i18n_server.MissionPollingThread.missionExec_n81i");
//				emailService.sendEmailForMission(mission.getMissionId(),"任务执行通知"," 任务执行开始");
				emailService.sendEmailForMission(mission.getMissionId(),"i18n_server.MissionPollingThread.missionExecNotice_n81i"," i18n_server.MissionPollingThread.missionExec_n81i");
				mission.setMissionState(2l);
				Common.getLoopMissionInfoMap().put(mission.getMissionId(), mission);
				
			}
			
			//-- 执行中操作（2）
			else if (mission.getMissionState().longValue() == 2l){		//未完成任务下发操作

				//-- 获取未下发的节点信息
				mrList = service.getMissionResultsForLoopTask(mission.getMissionId(),mission.getMissionType().intValue(),3);
				if(mrList == null || mrList.size() == 0){
					return ;
				}
				
			}
			
			//-- 准备撤销（5）
			else if (mission.getMissionState().longValue() == 5l){		//撤销任务下发操作 （获取当前操作周期或最后一个非初始化周期的轮次信息）
				logger.debug("周期任务 撤销操作");
				//-- 整理周期任务信息
				service.checkAndInitMissionCycle(mission);// 轮次校验和初始化
				List<LoopMissionRoundInfo> roundInfoList = Common.getLoopMissionRoundInfoList().get(mission.getMissionId());
				
				//-- 获取当前轮次信息并初始化当前轮次的任务结果信息
				if(roundInfoList !=null && roundInfoList.size()>0){
					
					//-- 获取当前周期轮次
					for (Iterator<LoopMissionRoundInfo> roundIte = roundInfoList.iterator(); roundIte.hasNext();) {
						roundInfo = roundIte.next();
						if(roundInfo.getEndTime().getTime() > System.currentTimeMillis()){
							break;
						}
					}
					//-- 旧轮次信息状态记录
					int oldRoundState = roundInfo.getMissionState()==null?0:roundInfo.getMissionState().intValue();
					
					//-- 将当前周期状态 变更为撤销准备周期
					service.updateLoopMissionState(roundInfo.getCurMissionId(),5);
					
					//-- 删除预置周期
					service.deleteLoopMissionPreset(mission.getMissionId(),roundInfo.getCurMissionId());
					
					//-- 更新缓存轮次信息
					service.updateLoopMissionRoundInfo(Common.getLoopMissionRoundInfoList(), mission.getMissionId());
					
					//-- 将撤销周期的结果数据进行状态初始化
					if(oldRoundState == 0){	//预置周期无结果
						
						service.initLoopMissionResults2(roundInfo.getCurMissionId(), task4.getNodeGroupsId(),task4.getNodeIpsId(),5);//初始化可下发的结果信息 5 撤销未下发
					
					}else if(oldRoundState>0){	//存在结果的周期 及2或3
						
						service.updateMissionResultsResult(roundInfo.getCurMissionId(), 4, 5);
					
					}
				}
				
				
				//-- 获取未下发的节点信息
				mrList = service.getMissionResultsForLoopTask(mission.getMissionId(),mission.getMissionType().intValue(),5);
				if(mrList == null || mrList.size() == 0){
					
					//-- 将周期状态变更为7 完成
					service.updateLoopMissionState(roundInfo.getCurMissionId(), 7);	//完成 任务轮次信息状态
					roundInfo.setMissionState(7l);	//更新缓存中的任务轮次信息状态
					return ;
				}else{
					
					//-- 将周期状态变更为6 开始执行
					service.updateLoopMissionState(roundInfo.getCurMissionId(), 6);	//更新 任务轮次信息状态
					roundInfo.setMissionState(6l);	//更新缓存中的任务轮次信息状态
				}
				//- 变更任务状态 5（准备撤销） -> 6（撤销执行）
//				service.updateMissionState(mission.getMissionId(), 6,format.format(System.currentTimeMillis())+" 任务撤销开始");
//				service.updateMissionState(mission.getMissionId(), 6,format.format(System.currentTimeMillis())+" Task revocation start");
				service.updateMissionState(mission.getMissionId(), 6,format.format(System.currentTimeMillis())+" i18n_server.MissionPollingThread.missionRevoke_n81i");
				mission.setMissionState(6l);
//				emailService.sendEmailForMission(mission.getMissionId(),"任务执行通知"," 任务撤销开始");
//				emailService.sendEmailForMission(mission.getMissionId(),"Task execution notice"," Task revocation start");
				emailService.sendEmailForMission(mission.getMissionId(),"i18n_server.MissionPollingThread.missionExecNotice_n81i"," i18n_server.MissionPollingThread.missionRevoke_n81i");
			}
			
			//-- 撤销中操作（6）
			else if (mission.getMissionState().longValue() == 6l){		//未完成任务下发操作

				//-- 获取未下发的节点信息
				mrList = service.getMissionResultsForLoopTask(mission.getMissionId(),mission.getMissionType().intValue(),5);
				if(mrList == null || mrList.size() == 0){
					service.updateLoopMissionComplete();
					return ;
				}
			}else{
				return;
			}
			
			//-- 将任务信息存入Map
			Common.getLoopMissionInfoMap().put(mission.getMissionId(), mission);

			
			task4.setNodeGroupsId(null);	//抹掉 节点组 信息
			LoopMissionRoundInfo roundInfo0 = null;
			
			if(mrList==null){
				return;
			}
			
			for (MissionResult result : mrList) {
				//-- 执行成功的结果 不再重复下发，跳过执行
				if(result.getResult() != null && result.getResult().longValue() == 0l){
					continue;
				}
				
				roundInfo0 = service.getRoundInfoFromCommonListByID(mission.getMissionId(), result.getMissionId());
					
				//从数据库中获取roundInfo
				if(roundInfo0 == null){
					roundInfo0 = service.getRoundInfoFromDBByID(result.getMissionId());
					
					//将roundInfo0保存到list中
					if(roundInfo0 != null){
						List<LoopMissionRoundInfo> list0 = Common.getLoopMissionRoundInfoList().get(mission.getMissionId());
						if(list0== null){
							list0 = new LinkedList<LoopMissionRoundInfo>();
							Common.getLoopMissionRoundInfoList().put(mission.getMissionId(),list0);
						}
						list0.add(0,roundInfo0);
					}else{
						continue;
					}
				}
				
				//-- 针对节点  进行文件推送
				if(mission.getMissionState().longValue() == 6l){
					Common.sendMissionInfo(mission,new NMSClientTask4Cancel(result.getNodeIp(),Constants.SSL_CLIENT_PORT,mission.getMissionId()),result.getUuid(), testNum,roundInfo0.getStartTime().getTime(),roundInfo0.getEndTime().getTime());
				}else if(mission.getMissionState().longValue() == 2l){
					Common.sendMissionInfo(mission, new NMSClientTask4(result.getNodeIp(),Constants.SSL_CLIENT_PORT,mission,task4,result),result.getUuid(), testNum,roundInfo0.getStartTime().getTime(),roundInfo0.getEndTime().getTime());
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}finally{
			if(dao!=null){
				dao.close();
				dao=null;
			}
		}
	}
	
	/**
	 * 升级任务任务执行
	 * @time Nov 11, 2011-6:55:33 PM
	 * @param mission
	 */
	public void forTask6(MissionStateTable mission) {
		CommonDao dao = null;			//数据库链接
		try {
			dao = new CommonDao();
			UpgradeService service = new UpgradeService(dao);	//专属Service
			
			//-- 任务结果查询和推送文件
			Task6 task6 = service.getTask6ById(mission.getMissionId());		//查询任务具体信息
			List<FileComment> fileCommentsList = service.getMissionFileInformations(mission.getMissionId());
			task6.setNodeGroupsId(null);	//抹掉 节点组 信息
			ArrayList<MissionResult> mrList = service.getMissionResultsForTask(mission.getMissionId(),mission.getMissionType().intValue());
			
			//-- 遍历节点结果集合
			if(task6.getVersion()!=null && task6.getVersion().longValue()!=0){	//逆向任务下发
				for (MissionResult result : mrList) {
					Common.sendMissionInfo(mission, new NMSClientTask6Cover(result.getNodeIp(),Constants.SSL_CLIENT_PORT,mission,task6,result,fileCommentsList),result.getUuid(), testNum,null , null);
				}
			}else{	//升级部署任务
				for (MissionResult result : mrList) {

					Common.sendMissionInfo(mission, new NMSClientTask6(result.getNodeIp(),Constants.SSL_CLIENT_PORT,mission,task6,result,fileCommentsList),result.getUuid(), testNum,null , null);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}finally{
			if(dao!=null){
				dao.close();
				dao=null;
			}
		}
	}
}