package com.nms.server.thread.socket;


import java.io.File;
import java.net.Socket;
import java.util.Date;
import java.util.LinkedList;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import net.sf.json.JSONArray;
import net.sf.json.JSONObject;

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

import com.nms.server.bean.MissionResult2;
import com.nms.server.common.Common;
import com.nms.server.common.Constants;
import com.nms.server.dao.CommonDao;
import com.nms.server.service.ChangeService;
import com.nms.server.service.CommonService;
import com.nms.server.service.UpgradeService;
import com.nms.server.thread.InitServerThread;
import com.nms.server.thread.ResetServerThread;
import com.nms.server.thread.alarmData.AlarmDataResoveThread;
import com.nms.server.thread.change.ChangeForActiveAlarmInfo;
import com.nms.server.thread.change.ChangeForCheckType;
import com.nms.server.thread.change.ChangeForNodeGroup;
import com.nms.server.thread.change.ChangeForSetInfo;
import com.nms.server.thread.change.ChangePluginScriptFile;
import com.nms.server.thread.detecData.NewDetecDataResoveThread;
import com.nms.server.thread.file.UnZipManagerThread;
import com.nms.server.thread.mission.LoadNewMissionThread;
import com.nms.server.thread.mission.MissionResultThread;
import com.nms.server.util.socket.SSLSocketCallable;


/**
 *  服务器通信进程  
 *  @date Nov 16, 2012 4:46:54 PM
 *	@author ZhangGang
 *
 */
public class SSLServer extends SSLSocketCallable{
	Logger logger = Logger.getLogger(SSLServer.class);

	public SSLServer(Socket socket) {
		super(socket);
	}
	
    /**
     * 业务类操作
     * @time Aug 28, 2011-8:41:37 PM
     * @param out
     * @param in
     * @throws Exception
     */
	protected Object toDo() throws Exception{
		CommonDao dao = null;
		try {
			String clientIp = socket.getInetAddress().getHostAddress();
//			Thread.currentThread().setName("接收通信："+clientIp);
			Thread.currentThread().setName("Receiving communication："+clientIp);
			
			//-- 获取业务操作命令
			String msg = receiveMessage();
			logger.info("通讯命令："+msg);
		
			/********* 与NMSClient通讯操作 *************************/
			//-- Agent端监测告警配置初始化
			if(SocketCMD.CLIENT_REQ_LOCAL_UUID.equalsIgnoreCase(msg)){
				dao = new CommonDao();
				CommonService service = new CommonService(dao);
				//- 校验IP是否存在
				//- 不存在，创建新节点并下发SEQ_ID(UUID)
				//-   存在，将旧的SEQ_ID(UUID)发送回去
				//新节点校验
				boolean newflag = service.newClientCheck(clientIp,null);
				//- 主动监测业务调整后,各监测线程每次执行监测时，即时查询数据库节点信息，无需手动变更，所以以下代码废弃
//				if(newflag){	//对新增节点初始化主动监测
//					List<SetInfo> setList = service.getDrivingSetInfo(Constants.DETEC_PING_STR, Common.getIpSeqIdMap().get(clientIp));
//					if(setList!=null && setList.size()>0){
//						for (Iterator setIte = setList.iterator(); setIte.hasNext();) {
//							SetInfo si = (SetInfo) setIte.next();
//							si.setSingleNodeId(Common.get);
//							logger.debug("监测启动>> "+si.getCheckTypeName());
//							new ThreadService(dao).updateSNMPThread(si);
//						}
//					}
//					setList.clear();
//					List<SetInfo> setList = service.getDrivingSetInfo(Constants.DETEC_NMSC_STR, Common.getIpSeqIdMap().get(clientIp));
//					if(setList!=null && setList.size()>0){
//						for (Iterator setIte = setList.iterator(); setIte.hasNext();) {
//							SetInfo si = (SetInfo) setIte.next();
//							
//							//- 设置名称
//							String processIden = StringUtils.isEmpty(si.getProcessIden())?"":si.getProcessIden();
//							
//							//-- 监测设置  线程名称 typeName_processIden
//							String runnableName =  si.getCheckTypeName()+"_"+processIden;
//							
//							//根据线程类别  线程对象
//							MonitorManagerThread managerThread = (MonitorManagerThread) Common.getMonitorRunnableFromMap(runnableName);
//							
//							if(managerThread!= null){
//								managerThread.addIp(new NodeModel(clientIp,0l));
//							}else{
//								logger.warn("主动监测线程："+runnableName+" 不存在");
//							}
//							
//						}
//					}
//				}
			    String returnMsg = Common.getIpSeqIdMap().get(clientIp)+ ":::" + clientIp;
			    sendMessage(returnMsg);
			    receiveMessage();
			}
			
			//-- Agent端监测告警配置初始化
			else if(SocketCMD.CLIENT_REQ_INIT_CONFIG.equalsIgnoreCase(msg)){
				dao = new CommonDao();
				ChangeService service = new ChangeService(dao);
				JSONObject object = service.agentInit(clientIp);
				sendMessage(object.toString());
				receiveMessage();
				
				//--清除该节点的所有变更事件
				service.deleteEventRecordBySeqId(Common.getUUIDByIp(clientIp));

				//--获取并发送脚本文件(初始化时发送脚)
				ChangePluginScriptFile.sendPluginFileWhenNcInit(clientIp);
			}
			//-- Agent端获取Server端系统日期
			else if(SocketCMD.CLIENT_REQ_SERVER_SYSTEMDATE.equalsIgnoreCase(msg)){
				sendMessage(new Date().getTime()+"");
				receiveMessage();
			}
			//-- 接收告警信息-----------------------------------------------------
			else if(SocketCMD.CLIENT_REQ_ALARM.equalsIgnoreCase(msg)){
				sendMessage(SUCCESS);
				String msg0 = receiveMessage();
				logger.info(msg0);
				sendMessage(SUCCESS);
				if(Constants.FLAG_ALARM_RESOVE==1){//启动了告警信息解析线程，才将接收的信息存入内容，否则不存，不然会将内存越撑越大
					//Ip type 进程标识  信息
					String [] alarms = msg0.split(Constants.COMMON_DATA_SPLIT);
					Common.addAlarmData(alarms);
				}
			}
			//-- 接收系统错误信息-----------------------------------------------------
			else if(SocketCMD.CLIENT_RQ_NC_ERROR_INFO.equalsIgnoreCase(msg)){
				sendMessage(SUCCESS);
				String msg0 = receiveMessage();
				logger.info(msg0);
				sendMessage(SUCCESS);
				//Ip type 进程标识  信息
				String [] alarms = msg0.split(Constants.COMMON_DATA_SPLIT);
				if(alarms.length==5){
					Common.addErrorInfo(alarms[0], alarms[1], alarms[2], new Date(Long.parseLong(alarms[3])), Long.parseLong(alarms[4]),null);
				}else if(alarms.length==6){
					Common.addErrorInfo(alarms[0], alarms[1], alarms[2], new Date(Long.parseLong(alarms[3])), Long.parseLong(alarms[4]),alarms[5]);
				}
				
			}
			//-- 接收监测数据文件到内存------------csv
			else if(SocketCMD.CLIENT_REQ_UPLOAD_DATAS.equalsIgnoreCase(msg)){
				sendMessage(SUCCESS);
				long sTime = System.currentTimeMillis();
				logger.debug("采集监测数据文件开始");
				LinkedList<byte []> bslist = receiveFileBytesByBath();
				logger.debug("采集监测数据文件结束 "+(System.currentTimeMillis()-sTime)+" ms");
				if(bslist!=null && bslist.size()>0){
					Common.addAllDeteDataList(bslist);
				}else{
					logger.debug("接收 监测数据 文件数:0");
				}
				sendMessage(SUCCESS);
			}
			//-----------------------------------------------------------------------------------------
			//-- 接收返回的任务结果
			else if(SocketCMD.CLIENT_REQ_TASK_RESULT.equalsIgnoreCase(msg)){
				sendMessage(SUCCESS);
				String msg0 = receiveMessage();
				logger.info(msg0);
				sendMessage(SUCCESS);
				String [] result = msg0.split(Constants.COMMON_DATA_SPLIT);
				MissionResult2 result2 = Common.resoveMissionResult(result);
				Common.addResultToResultList(result2);
			}
			
			//-- Agent请求任务信息
			else if(SocketCMD.CLIENT_REQ_INIT_TASK.equalsIgnoreCase(msg)){
				dao = new CommonDao();
				UpgradeService uService = new UpgradeService(dao);
				sendMessage(SUCCESS);
				String uuid = receiveMessage();
				JSONArray array = new JSONArray();
				if(StringUtils.isNotEmpty(uuid)){
					array =  uService.getMissionForAgentStart(Long.parseLong(uuid));
				}
				sendMessage(array.toString());
				receiveMessage();
			}
			
			//-- Agent端，通信握手
			else if(SocketCMD.CLIENT_REQ_HAND_SHAKE.equalsIgnoreCase(msg)){
				sendMessage(SUCCESS);
//				if(!Constants.INIT_FLAG){
//					Common.service.submit(new InitServerThread());
//				}
			}
			
			//-- 接收任务4“启动”的回传文件
			else if(SocketCMD.CLIENT_REQ_TASK_RETURNFILE.equalsIgnoreCase(msg)){
				
				sendMessage(SUCCESS);
				
				String msg0 = receiveMessage();
				sendMessage(SUCCESS);
				
				logger.info(msg0);
				String [] result = msg0.split(Constants.COMMON_DATA_SPLIT);
				MissionResult2 result2 = Common.resoveMissionResult(result);
				String fileName = receiveMessage();
				sendMessage(SUCCESS);
				logger.debug("MISSION_FILE_UPLOAD_DIR"+Constants.MISSION_FILE_UPLOAD_DIR);
				
				this.bpReceiveFile(Constants.MISSION_FILE_UPLOAD_DIR+fileName);

				sendMessage(SUCCESS);
				logger.debug("MISSION_FILE_UPLOAD_DIR"+Constants.MISSION_FILE_UPLOAD_DIR+fileName);
				
				result2.setFileInfo(fileName);
				Common.addResultToResultList(result2);
			}

			//-- 服务端升级
			else if(SocketCMD.CLIENT_REQ_SERVER_UPGRADE.equalsIgnoreCase(msg)){
				//清理缓存操作并"关闭"通讯
				try {
					//-- 第一步 关闭通讯
					Common.SERVER_UN_UPGRADE_FLAG = false;
					
					//-- 第二步 通讯恢复“定时”:当NC杀死DC进程不成功时，DC还可以按照旧代码处理工作（监测，任务等）
					Common.scheduled.schedule(new Runnable(){
						public void run() {
							Common.SERVER_UN_UPGRADE_FLAG = true;
							Common.service.submit(new InitServerThread());
						}},Constants.MISSION_UPGRADE_DAILY,TimeUnit.SECONDS);
					//-- 第二步 清理缓存 detection & alarm & results
					
					//-- 关闭线程 清理 监测数据
					if(Common.getFutureMap().get(Constants.DETEC_DATA_RESOVE) != null){
						Common.getFutureMap().get(Constants.DETEC_DATA_RESOVE).cancel(true);
					}
					LinkedList<byte []> dsbList = Common.getDeteDataList();
					Future<?> detecFuture = Common.scheduled.submit(new NewDetecDataResoveThread("Server升级前_监测数据缓存入库1",dsbList));
					Common.chengeDeteDataFlag();
					LinkedList<byte []> dsbList2 = Common.getDeteDataList();
					Future<?> detecFuture2 = Common.scheduled.submit(new NewDetecDataResoveThread("Server升级前_监测数据缓存入库2",dsbList2));
					
					//-- 关闭线程 清理 告警数据
					if(Common.getFutureMap().get(Constants.ALARM_DATA_RESOVE) != null){
						Common.getFutureMap().get(Constants.ALARM_DATA_RESOVE).cancel(true);
					}
					LinkedList<String []> adList = Common.getAlarmDataList();
					Future<?> alarmFuture = Common.scheduled.submit(new AlarmDataResoveThread("Server升级前_告警数据缓存入库1",adList));
					Common.changeAlarmDataFlag();
					LinkedList<String []> adList2 = Common.getAlarmDataList();
					Future<?> alarmFuture2 = Common.scheduled.submit(new AlarmDataResoveThread("Server升级前_告警数据缓存入库2",adList2));
					
					//-- 关闭线程 清理 任务结果数据
					if(Common.getFutureMap().get(Constants.MISSION_RESULT_RESOVE) != null){
						Common.getFutureMap().get(Constants.MISSION_RESULT_RESOVE).cancel(true);
					}
					Future<?> mrFuture = Common.scheduled.submit(new MissionResultThread("Server升级前_任务结果缓存入库"));
					try {
						detecFuture.get(Constants.MISSION_UPGRADE_DAILY, TimeUnit.SECONDS);
						detecFuture2.get(Constants.MISSION_UPGRADE_DAILY, TimeUnit.SECONDS);
						
						alarmFuture.get(Constants.MISSION_UPGRADE_DAILY, TimeUnit.SECONDS);
						alarmFuture2.get(Constants.MISSION_UPGRADE_DAILY, TimeUnit.SECONDS);
						
						mrFuture.get(Constants.MISSION_UPGRADE_DAILY, TimeUnit.SECONDS);
					} catch (Exception e) {
						logger.error("DC cleaning data timeout, cache data stored in the hard disk", e);
						Common.saveDataInMemToDisk();
						logger.warn("The cached data is cleaned up and the upgrade is confirmed");
					}finally{
						//-- 最后一步返回success
						sendMessage(SUCCESS);
					}
				} catch (Exception e) {
					logger.error("Server upgrade preparation work failed!", e);
					//-- 最后一步返回success
					sendMessage(FAIL);
				}
			}
			
			//-- 数据压缩文件接收--zip
			else if(msg!=null && msg.startsWith(SocketCMD.CLIENT_REQ_BP_UPLOAD_FIFE)){
				this.sendMessage(SUCCESS);
				String[] cmds = msg.split(":");
				String uploadPath = Constants.ZIP_FILE_DETECT_DATA_DIR;
				if(cmds.length>2){
					if(cmds[2].equals(SocketCMD.SUFFIX_BP_TYPE_DETECT_DATA)){	//download MissionFile
						uploadPath = Constants.ZIP_FILE_DETECT_DATA_DIR;
					}else if(cmds[2].equals(SocketCMD.SUFFIX_BP_TYPE_TASK_RESULT)){	//download snmp jarFile
						uploadPath = Constants.ZIP_FILE_TASK_RESULT_DIR;
					}else if(cmds[2].equals(SocketCMD.SUFFIX_BP_TYPE_TASK_RETURN)){	//download snmp mibFile
						uploadPath = Constants.ZIP_FILE_TASK_RETURN_DIR;
					}
				}
				String fileName = receiveMessage();
				this.sendMessage(SUCCESS);
				bpReceiveFile(uploadPath+File.separator+fileName);
				this.sendMessage(SUCCESS);
				
				//启动并注册 解压文件线程
				Future<?> future = Common.getFutureMap().get(Constants.UNZIP_FILE_MANAGER);
				if(future == null){
					future = Common.scheduled.scheduleWithFixedDelay(new UnZipManagerThread(), 0, 30, TimeUnit.SECONDS);
					Common.registRunnable(Constants.UNZIP_FILE_MANAGER, future);
				}
			}
			//-- Agent申请Node参数变更（Ip systemType）
			else if(SocketCMD.CLIENT_AGENT_CHANGE.equalsIgnoreCase(msg)){
				dao = new CommonDao();
				CommonService service = new CommonService(dao);
				sendMessage(SUCCESS);
				String paramsStr = receiveMessage();
//				pl("paramsStr"+paramsStr);
//				UUID$@$操作系统类型（1、2）$@$IP
				String[] resultInfo = new String[2];
				if(StringUtils.isNotEmpty(paramsStr)){
					String []  params = paramsStr.split(Constants.COMMON_DATA_SPLIT);
					resultInfo =  service.checkAndUpdateNodeParams(Long.parseLong(params[0]),clientIp,Integer.parseInt(params[1]));
				}
				sendMessage(resultInfo[0]+Constants.COMMON_DATA_POINT+resultInfo[1]);
				receiveMessage();
			}
			//-- 初始化任务信息
			else if("byte:initTask".equalsIgnoreCase(msg)){
				dao = new CommonDao();
				UpgradeService uService = new UpgradeService(dao);
				sendMessage(SUCCESS);
				String uuid = receiveMessage();
				JSONArray array = new JSONArray();
				if(StringUtils.isNotEmpty(uuid)){
					array =  uService.getMissionForAgentStart(Long.parseLong(uuid));
				}
				sendMessage(array.toString());
				receiveMessage();
			}

			//-- 证书接收操作
			else if(msg!=null && msg.toLowerCase().startsWith("byte:cert")){
				String aliasName = msg.split(":")[2];
				sendMessage(SUCCESS);
				String cerDir = Constants.SYSTEM_PATH+File.separator+Constants.SSL_DIR+File.separator;
				String certName = cerDir + clientIp+".cer";
				String storePath = cerDir + Constants.SSL_SERVER_TRUST;
				receiveFile(certName);
				SSLCertOper.importCertToStore(aliasName, storePath, certName);
				sendMessage(SUCCESS);
			}
			
			/********* 与NMSWeb通讯操作 *************************/
			//-- 判断内容，执行操作
			if (SocketCMD.WEB_NOTICE_SET_INFO_ALERT.equalsIgnoreCase(msg)) {
				/**监测设置信息变更或时，进行此操作
				 * 影响：新旧设置信息在有效性、监测范围、上均有可能变化，
				 * 需要详细比较新旧信息后，再进行下发
				 * 有效性变化：通知最新
				 * 监测范围变化：先通知旧设置信息取消监测
				 * 			  再通知新节点信息启动监测
				 * 监测内容变化：通知最新
				 * （未完成）
				 * */
				//-- 通讯接收信息
				
				sendMessage(SUCCESS);
				String datas = receiveMessage(); //JsonObject 对象
				logger.debug(""+datas);
				sendMessage(SUCCESS);
				
				//-- 处理通讯信息
				
				Common.service.execute(new ChangeForSetInfo(null,datas));
			}
			
			else if (SocketCMD.SEND_PLUGIN_SCRIPT_FILE.equalsIgnoreCase(msg)) {
				/**
				 * 下发第三方监测脚本
				 * */
				//-- 通讯接收信息
				sendMessage(SUCCESS);
				File tempPluginDir = ChangePluginScriptFile.getNewTempPluginDirectory();
				this.bpReceiveFileByBath(tempPluginDir.getCanonicalPath());
				sendMessage(SUCCESS);
				//-- 处理通讯信息
				
				Common.service.execute(new ChangePluginScriptFile(tempPluginDir));
			}
			
			else if (SocketCMD.WEB_NOTICE_CHECK_TYPE_ALERT.equalsIgnoreCase(msg)) {	//Server端更新缓存信息方法
				/**告警、库表信息变更时，进行此操作
				 * 影响：NMSServer端解析需要更新
				 * 对Agent端运行无影响及操作。
				 * （已完成）
				 * */
				sendMessage(SUCCESS);
				String datas = receiveMessage(); //setInfoId:checkWay:checkState
				pl(datas);
				String[] data = datas.split(":");
				sendMessage(SUCCESS);
				
				Common.service.execute(new ChangeForCheckType(null,data));
			}
			else if (SocketCMD.WEB_NOTICE_NODE_AND_GROUP_STATE_ALERT.equalsIgnoreCase(msg)) {
				/**节点或节点组新增或有效性变化时，进行此通信
				 * 影响：仅指状态变化，范围不变
				 * 重新查询节点或节点组涉及的监测设置信息，再次下发即可
				 * （待测试）
				 * */
				//-- 通讯开始 并接收数据信息
				sendMessage(SUCCESS);
				
				String datas = receiveMessage(); //groupId:nodeId   [节点组Id:节点ID 节点Id为空时，即为节点组信息 ]
				logger.debug("NodeGroup & Node Info: "+datas);
				sendMessage(SUCCESS);
				
				//-- 通信结束，数据解析
				String[] data = datas==null?null:datas.split(":");
				//-- 通过新线程 变更影响范围内的相关信息
				if(data!=null && data.length>0 && StringUtils.isNotEmpty(data[0])){
					Common.service.submit(new ChangeForNodeGroup(null,data) );
				}else{
					logger.warn("Web end “NodeOrGroupEdit” data format is incorrect");
				}
			}
			else if (SocketCMD.WEB_NOTICE_MISSION_INFO_ALERT.equalsIgnoreCase(msg)) {//任务信息变更，包括创建任务后的任务下发
				sendMessage(SUCCESS);
				
				String datas = receiveMessage(); //missionId:osLimits
				logger.debug("upgrade Message: "+datas);
				String[] data = datas.split(":");
				//判断是否需要即时初始化
				//if(data.length>1 && "1".equals(data[1]) ){
				//	 Future<?> future = Common.service.submit(new LoadNewMissionThread(Long.parseLong(data[0].trim())));
				//	 future.get();
					 
				//}else{
					sendMessage(SUCCESS);
				//}
				
				if(data!=null && data.length>0 && StringUtils.isNotEmpty(data[0])){
					Common.service.submit(new LoadNewMissionThread(Long.parseLong(data[0].trim())));
				}else{
					logger.warn("Web terminal “missionInfo” task send empty data communication");
				}
			}
			else if (SocketCMD.WEB_NOTICE_DATACONTROLLER.equalsIgnoreCase(msg)) {
				sendMessage(SUCCESS);
				
				String datas = receiveMessage(); //int  1 更新自身缓存信息
				logger.debug("upgrade Message: "+datas);
				sendMessage(SUCCESS);
				if(StringUtils.isNotBlank(datas)){
					Common.service.submit(new ResetServerThread(Integer.parseInt(datas.trim())));
				}else{
					logger.warn("Web terminal “"+SocketCMD.WEB_NOTICE_DATACONTROLLER+"”task send empty data communication");
				}
			}
			else if (SocketCMD.WEB_NOTICE_ACTIVE_ALARM_START_ALERT.equalsIgnoreCase(msg)) {//NC端是否报主动告警
				sendMessage(SUCCESS);
				
				String datas = receiveMessage();
				logger.debug("upgrade Message: "+datas);
				sendMessage(SUCCESS);
				if(StringUtils.isNotBlank(datas)){
					Common.service.submit(new ChangeForActiveAlarmInfo(null,datas));
				}else{
					logger.warn("Web terminal “"+SocketCMD.WEB_NOTICE_ACTIVE_ALARM_START_ALERT+"”send empty data communication");
				}
			}
			else if (SocketCMD.WEB_UPDATE_CONFIG.equalsIgnoreCase(msg)) {
				sendMessage(SUCCESS);
				
				String datas = receiveMessage(); //int  1 更新自身缓存信息
				logger.debug("upgrade Message: "+datas);
				sendMessage(SUCCESS);
				if(StringUtils.isNotBlank(datas)){
					Common.service.submit(new ResetServerThread(Integer.parseInt(datas.trim())));
				}else{
					logger.warn("Web terminal “"+SocketCMD.WEB_NOTICE_DATACONTROLLER+"”task send empty data communication");
				}
			}else if(SocketCMD.WEB_NOTICE_INSERT_DETECT.equalsIgnoreCase(msg)){
				/**
				 * 通知监测数据入库
				 */
				int size = -1;
				Common.setLastWebHandshake(System.currentTimeMillis());
				try {
					logger.debug("web :"+ clientIp +" 通知监测数据入库");
					if(Constants.DETECT_INSERT_MODE == 1 || Constants.DETECT_INSERT_MODE == 3){
						//- 非升级操作判断
						if(Common.SERVER_UN_UPGRADE_FLAG){
							Common.chengeDeteDataFlag();	//变更数据存放集合
							LinkedList<byte []> dsbList =  Common.getDeteDataList();// 获取非存放状态的数据集合
							size = dsbList.size();
							if(dsbList != null && size>0){
								//-- 获取线程执行 需进行主动告警和邮件通知等相关操作，待考虑
								Future future = Common.dataResoveService.submit(new NewDetecDataResoveThread("监测数据解析执行线程",dsbList));
								//注册
								Common.getFutureMap().put(Constants.DETEC_DATA_RESOVE, future);
							}else{
								logger.info("获取到的集合中数据数量为0，无需启动解析线程");
							}
						}else{
							logger.debug("升级中，暂不入库");
						}
					}else{
						size = Constants.DETECT_INSERT_MODE -6;//2主动入库 -4，4上传web -2 
						logger.debug("DETECT_INSERT_MODE = " + Constants.DETECT_INSERT_MODE +" 不做处理");
					}
				} catch (Exception e) {
					logger.error("",e);
				}finally{
					sendMessage(size+"");
				}
			}else if(SocketCMD.RELEASE_DETECT.equalsIgnoreCase(msg)){
				/**
				 * 通知 关闭监测数据入库
				 */
				boolean cancel = false;
				try {
					Future<?> future = Common.getFutureMap().get(Constants.DETEC_DATA_RESOVE);
					boolean done = future.isDone();
					logger.debug("监测数据解析执行线程 isDone:" + done);
					cancel = future.cancel(true);
					logger.debug("监测数据解析执行线程 cancel : " + cancel);
					
				} catch (Exception e) {
					logger.error("",e);
				}finally{
					sendMessage(cancel+"");
				}
			}
			
		} catch (Exception e) {
			logger.error("Communicating exception",e);
		}finally{
			if(dao!=null){
				dao.close();
				dao=null;
			}
			logger.debug("通讯结束");
		}
		return null;
    }
}
