locprefixioprocess.java

中国移动定位引擎的客户端

	
		log.info("新的LocInfoMgr连接" + sc.socket().getRemoteSocketAddress());

		int loginStatus = LoginRsp.SUCC;
		ByteBuffer readBuf = receiveBindReq(sc);
		if (readBuf.position() < LeMessage.HEADER_LENGTH) {
			loginStatus = LoginRsp.ERR_TimeOut;
			log.error("接收包登录超时" + sc);
		}
		// 判断登录消息包是否超时
		if (loginStatus != LoginRsp.SUCC) {
			processInvaildLogin(sc, loginStatus);
			return;
		}

		readBuf.flip();

		
		Login login = new Login();
		try {
			login.readMsg(readBuf);
		} catch (ProtocolException mpe) {
			loginStatus = LoginRsp.ERR_Parse;
			log.error("解析登录包，错误", mpe);
		}
		log.debug(login.getSimpleInfo());

		// 判断消息解析否成功
		if (loginStatus != LoginRsp.SUCC) {
			processInvaildLogin(sc, loginStatus);
			return;
		}
		LocInfoMgr locInfoMgr = new LocInfoMgr();
		try {
			LoginInfoDao lgDao=new LoginInfoDao();
			boolean res=lgDao.getLocInfoMgr(login.getUserId(), login.getPassword(), login.getOrgSystemID());
			if (res ==false) {
				loginStatus = LoginRsp.ERR_Passwd;
				log.warn("位置信息管理模块登录时，用户名，密码或源模块号不匹配" + sc);
			}
		} catch (SQLException sqle) {
			loginStatus = LoginRsp.ERR_DB;
			log.error("位置信息管理模块登录时，数据库检验异常", sqle);
		}

		// 测试的时候使用，记录消息包的数量
//		MessageCount msgCon=new MessageCount(login.getOrgSystemID()+"",login.getCommandId());
//		msgCon.incrementReceive(login.getOrgSystemID(),login.getCommandId());

		// 判断实体信息是否存在
		if (loginStatus != LoginRsp.SUCC) {
			processInvaildLogin(sc, loginStatus);
			return;
		}
		Connection existConn = (Connection) clientConns.get(login.getOrgSystemID());
		if (existConn != null) {
			if (existConn.isConnected()) {
				loginStatus = LoginRsp.ERR_RELOGIN;
				log.warn("位置信息管理模块登录时，重复登录" + sc);
			} else {
				// 处理断开的连接
				existConn.destroy();
				clientConns.delete(login.getOrgSystemID());
			}
		}
		// 判断是否重复登录
		if (loginStatus != LoginRsp.SUCC) {
			processInvaildLogin(sc, loginStatus);
			return;
		}

		/** 登录成功，后加入到mspConns集合，保存起该连接 */
		locInfoMgr.setEmseNo("位置管理"+SysUtil.getModuleId(login.getOrgSystemID())+"号机");
		LocInfoMgrConfig config = new LocInfoMgrConfig();
		//这一段有问题，config改了之后，这里也要改
		config.setEmseInfo(locInfoMgr);
		config.setConnID(login.getOrgSystemID());
		//end of 这一段有问题
		Connection conn = ConnectionFactory.createLocPrefixConnection(config,
				getIoProcessor(), getEventDispatcher(), LocPrefixEncoder, this);
		conn.setConfig(config);
		conn.setConnTypeIsServer(true);
		try {
			conn.start(sc);
			conn.getConfig().setConnStatus(Connection.LOGIN_ON);
			LoginRsp loginrsp = new LoginRsp();
			loginrsp.setResult(LoginRsp.SUCC);
			if (conn.sendMessage(loginrsp)) {
				log.info("位置信息管理模块登录成功：" + conn);
			}
			clientConns.put(login.getOrgSystemID(), conn);
		} catch (IOException ioe) {
			conn.close();
			log.error("位置信息管理模块登录时出错", ioe);
		}
	}

	/**
	 * 处理MSP网关登录到MSP失败的方法
	 * 
	 * @param sc
	 * @param status
	 */
	private void processInvaildLogin(SocketChannel sc, int status) {
		LoginRsp loginRsp = new LoginRsp();
		loginRsp.setSequenceId(LbpUtil.getNextSeqNum());
		loginRsp.setResult(status);
		try {
			super.write(sc, loginRsp);
		} catch (IOException ioe) {
			log.error("回登录应答包出错:" + sc + loginRsp, ioe);
		} finally {
			closeSocket(sc);
		}
	}

	/**
	 * 线程处理方法，定时调用该方法,维护用
	 */
	public void workProcess() {
	/*	// 检查数据库更新时间是否到
		if (checkSynchEmseTime() || !initialize) {
			initialize = true;
			this.syncEmseInfo();
		}*/
		// 检查外部实体的连接状态
		if (clientConns != null && clientConns.getSize() > 0) {
			checkLocInfoMgrConnsStatus(clientConns);
		}
/*		// 检查外部实体的连接状态
		if (serverConns != null && serverConns.getSize() > 0) {
			checkLocInfoMgrConnsStatus(serverConns);
		}*/

/*		// 链路状态定时通知
		if ((System.currentTimeMillis() - lastLinkNotifyTime) > getLinkNotifyTime()) {
			super.notifyLinkStatusToAllMsp(clientConns);
			super.notifyLinkStatusToAllMsp(serverConns);
			lastLinkNotifyTime = System.currentTimeMillis();
		}*/
	}

	/**
	 * 检查外部实体的链路状态
	 */
	private void checkLocInfoMgrConnsStatus(SimpleHashtable emseConns) {
		/*
		 * 遍历KEY
		 */
		Iterator key = emseConns.keyIterator();
		while (key.hasNext()) {
			int keyIndex = ((Integer) key.next()).intValue();
			Connection conn = (Connection) emseConns.get(keyIndex);
			if (conn == null) {
				continue;
			}
			// 外部实体的状态
			int emseStatus = conn.getConfig().getEmseInfo().getEmseStatus();
			// 判断外部实体状态是否可用，如果不可用，关闭连接
			if (emseStatus != LocInfoMgrConfig.LOCINFOMGR_STATUS_VALID) {
				if (conn.isConnected()) {
					// notifyLinkStatusToMsp(conn, false);
					processDisconnect(conn);
				}
				emseConns.delete(conn.getConnID());
				continue;
			}

			// 判断连接状态，用发送链路检测包的方式
			if (!checkIsConnect(conn)) {
				if (conn.connTypeIsServer()) {
					conn.destroy();
					emseConns.delete(conn.getConnID());
				} else {
					// 如果是做为客户端的链路，必须重连
					processEmseDisconnect(conn);
				}
			}
		}
	}

	/**
	 * 处理断开的连接
	 * 
	 * @param conn
	 */
	private void processEmseDisconnect(Connection conn) {

		if (conn.isConnected()) {
			processDisconnect(conn);
		}
		processReConnect(conn);
	}

	/**
	 * 判断数据库更新时间是否到
	 * 
	 * @return boolean
	 */
	protected boolean checkSynchEmseTime() {
		return ((System.currentTimeMillis() - lastSynchEmseInfoTime) > LocInfoMgrConfig
				.getSynchEmseInfoTime());
	}

	/**
	 * 异常处理方法，须进一步完善
	 * 
	 * @param conn
	 *            连接对象
	 * @param throwable
	 *            异常对象
	 */
	public void exceptionCaught(Connection conn, Throwable throwable) {
		log.error(conn, throwable);
	}
	/**
	 * 处理获取用户位置的请求，第一阶段暂不支持，返回一个错误的应答
	 * @param conn
	 * @param message
	 * @return true:返回应答成功
	 */
	public boolean processGetUserLoc(Connection conn, LbpMessage message){
		GetUserLoc get_user_loc=(GetUserLoc)message;
		GetUserLocRsp get_user_loc_rsp=new GetUserLocRsp();
		get_user_loc_rsp.setSequenceId(get_user_loc.getSequenceId());
		get_user_loc_rsp.setOrgSystemID(get_user_loc.getDstSystemID());
		get_user_loc_rsp.setDstSystemID(get_user_loc.getOrgSystemID());
		get_user_loc_rsp.setResult(1);
		TLVTable xml=new TLVTable();
		xml.set(Tag.ShapeDesc, "暂时不支持获取xml位置信息的功能");
		get_user_loc_rsp.setXML(xml);
		return conn.sendMessage(get_user_loc_rsp);
		}
		
	/**
	 * 处理增加用户的请求，第一阶段暂不支持，返回一个错误的应答
	 * @param conn
	 * @param message
	 * @return true:返回应答成功
	 */
	public boolean processAddLocUser(Connection conn, LbpMessage message){
		AddLocUser add_loc_user=(AddLocUser)message;
		AddLocUserRsp add_loc_user_rsp=new AddLocUserRsp();
		add_loc_user_rsp.setSequenceId(add_loc_user.getSequenceId());
		add_loc_user_rsp.setOrgSystemID(add_loc_user.getDstSystemID());
		add_loc_user_rsp.setDstSystemID(add_loc_user.getOrgSystemID());
		add_loc_user_rsp.setResult(2);//0:成功	1:失败要重发	2:失败不要重发
		add_loc_user_rsp.setIsdn(add_loc_user.getIsdn());
		return conn.sendMessage(add_loc_user_rsp);
		}
		
	/**
	 * 处理删除用户的请求，第一阶段暂不支持，返回一个错误的应答
	 * @param conn
	 * @param message
	 * @return true:返回应答成功
	 */	
	public boolean processDelLocUser(Connection conn, LbpMessage message){
		DelLocUser del_loc_user=(DelLocUser)message;
		DelLocUserRsp del_loc_user_rsp=new DelLocUserRsp();
		del_loc_user_rsp.setSequenceId(del_loc_user.getSequenceId());
		del_loc_user_rsp.setOrgSystemID(del_loc_user.getDstSystemID());
		del_loc_user_rsp.setDstSystemID(del_loc_user.getOrgSystemID());
		del_loc_user_rsp.setResult(2);//0:成功	1:失败要重发	2:失败不要重发
		del_loc_user_rsp.setIsdn(del_loc_user.getIsdn());
		return conn.sendMessage(del_loc_user_rsp);
		}
		
	/**
	 * 处理标准定位的请求，第一阶段暂不支持，返回一个错误的应答
	 * @param conn
	 * @param message
	 * @return true:返回应答成功
	 */		
		public boolean processUserTrig(Connection conn, LbpMessage message){
			UserTrig std_loc=(UserTrig)message;
			UserTrigRsp std_loc_rsp=new UserTrigRsp();
			std_loc_rsp.setSequenceId(std_loc.getSequenceId());
			std_loc_rsp.setOrgSystemID(std_loc.getDstSystemID());
			std_loc_rsp.setDstSystemID(std_loc.getOrgSystemID());
			std_loc_rsp.setResult(1);
			return conn.sendMessage(std_loc_rsp);
		}
		
		/**
		 * 处理周期性定位的请求
		 * @param conn
		 * @param message
		 * @return true:返回应答成功
		 */			
		public void processCycLoc(Connection conn, LbpMessage message){
			CycLoc cyc_loc=(CycLoc)message;
			CycLocRsp cyc_loc_rsp=new CycLocRsp();
			cyc_loc_rsp.setSequenceId(cyc_loc.getSequenceId());
			cyc_loc_rsp.setOrgSystemID(cyc_loc.getDstSystemID());
			cyc_loc_rsp.setDstSystemID(cyc_loc.getOrgSystemID());
			cyc_loc_rsp.setResult(0);
			//先返回一个应答
			this.sendToLocInfoMgr(cyc_loc_rsp,conn);
			SagCycLoc sagcycloc=new SagCycLoc(cyc_loc);
			try{
//				sagcycloc.bulidSagCycLoc();
               }catch (Exception e){
                log.error("向定位引擎发起周期性请求时，出现异常"+e);
               }
		}
		
		/**
		 * 处理周期性定位的终止请求
		 * @param conn
		 * @param message
		 * @return true:返回应答成功
		 */			
		public void processStopCycLoc(Connection conn, LbpMessage message){
			StopCycLoc cyc_loc_stop=(StopCycLoc)message;
			StopCycLocRsp cyc_loc_stop_rsp=new StopCycLocRsp();
			cyc_loc_stop_rsp.setSequenceId(cyc_loc_stop.getSequenceId());
			cyc_loc_stop_rsp.setOrgSystemID(cyc_loc_stop.getDstSystemID());
			cyc_loc_stop_rsp.setDstSystemID(cyc_loc_stop.getOrgSystemID());
			cyc_loc_stop_rsp.setResult(0);
			conn.sendMessage(cyc_loc_stop_rsp);
			//向isag发出webservice请求
			SagStopCycLoc stopsagcycloc=new SagStopCycLoc(cyc_loc_stop);
			try{
//				stopsagcycloc.bulidSagStopCycLoc();
               }catch (Exception e){
                log.error("向定位引擎发起周期性请求时，出现异常"+e);
               }
		
			
		}
		
		/**
		 * 处理区域触发的请求，第一阶段暂不支持，返回一个错误的应答
		 * @param conn
		 * @param message
		 * @return true:返回应答成功
		 */	
		public boolean processTriArea(Connection conn, LbpMessage message){
			TriArea tri_area=(TriArea)message;
			TriAreaRsp tri_area_rsp=new TriAreaRsp();
			tri_area_rsp.setSequenceId(tri_area.getSequenceId());
			tri_area_rsp.setOrgSystemID(tri_area.getDstSystemID());
			tri_area_rsp.setDstSystemID(tri_area.getOrgSystemID());
			tri_area_rsp.setResult((char)1);
			return conn.sendMessage(tri_area_rsp);
		}
		
		/**
		 * 处理区域触发的终止请求，第一阶段暂不支持，返回一个错误的应答
		 * @param conn
		 * @param message
		 * @return true:返回应答成功
		 */	
		public boolean processTriAreaStop(Connection conn, LbpMessage message){
			TriAreaStop tri_area_stop=(TriAreaStop)message;
			TriAreaStopRsp tri_area_stop_rsp=new TriAreaStopRsp();
			tri_area_stop_rsp.setSequenceId(tri_area_stop.getSequenceId());
			tri_area_stop_rsp.setOrgSystemID(tri_area_stop.getDstSystemID());
			tri_area_stop_rsp.setDstSystemID(tri_area_stop.getOrgSystemID());
			tri_area_stop_rsp.setResult((char)0);
			return conn.sendMessage(tri_area_stop_rsp);
		}
		
		/**
		 * 处理顶点测距的请求，第一阶段暂不支持，返回一个错误的应答
		 * @param conn
		 * @param message
		 * @return true:返回应答成功
		 */	
		public boolean processDstan(Connection conn, LbpMessage message){
			Dstan dstan=(Dstan)message;
			DstanRsp dstan_rsp=new DstanRsp();
			dstan_rsp.setSequenceId(dstan.getSequenceId());
			dstan_rsp.setOrgSystemID(dstan.getDstSystemID());
			dstan_rsp.setDstSystemID(dstan.getOrgSystemID());
			dstan_rsp.setResult((char)1);
			return conn.sendMessage(dstan_rsp);
		}
		

}