cmppconnection.java

基于中国网通cngp2.0开发的一个java网关源代码。大家多多公开SP类的网关程序哦

package cn.netjava.cngpclient.connection;

import java.io.*;
import java.net.*;
import java.util.*;
import java.util.logging.*;

import cn.netjava.cngpclient.log.LogManage;
import cn.netjava.cngpclient.publics.*;

/**
 * 网络通信连结对象
 * 中国网通短信网关客户端--蓝杰实训项目
 * @author www.NetJava.cn
 *
 */
public class CmppConnection extends Thread {

	public CmppConnection(String ip, int port, byte bindtype, String spip) {
		myip = new String(ip);
		myport = port;
		myBindType = bindtype;
		mystate = ComConstants.NOTCONNECTED;
		myLastVersion = 16;
		myConnType = CmppConstants.CLIENT;
		spIP = spip;
		try {
			vcmppMsg1 = new Vector(ComConstants.CmppVECTOR_NUM / 2, 100);
			vcmppMsg2 = new Vector(ComConstants.CmppVECTOR_NUM / 2, 100);
			vCmppSubmitMsg = new Vector(ComConstants.CmppVECTOR_NUM, 100);
			vcmppMsgFlag = 1;
			handleMsgThread();
		} catch (Exception ex) {
			LogManage.ins().myLog.severe(" CNGP  Create Connetion "
					+ ex.toString());
		}
	}

	public boolean connect() {
		try {
			InetSocketAddress myServer = new InetSocketAddress(IsmgInfo.ismgIp,
					IsmgInfo.ismgMtPort);
			InetSocketAddress myClient = new InetSocketAddress(spIP, IsmgInfo
					.getClientPort());
			mySocket = new Socket(myServer.getAddress(), myServer.getPort(),
					myClient.getAddress(), myClient.getPort());
			mySocket.setSoTimeout(ComConstants.CmppRCVSOCK_TIMEOUT);
			myinputstream = mySocket.getInputStream();
			myoutputstream = mySocket.getOutputStream();
			dataInStream = new DataInputStream(myinputstream);
			LogManage.ins().myLog.logp(Level.INFO, "", "", "connect serverIP: "
					+ myServer.getAddress() + "  serverPort: "
					+ myServer.getPort());
			return true;
		} catch (Exception e) {
			disconnect();
			mystate = ComConstants.NOTCONNECTED;
			LogManage.ins().myLog.severe("Cmpp Connected ISMG fail: " + e);
			return false;
		}
	}

	public void send(ShortMsgCmpp sm) {
		try {
			// debugMsg(sm);
			byte[] bb = sm.getPackage();

			if (bb == null) {
				// LogManage.ins().myLog .severe(" null package.");
				return;
			}
			// if ( (bb.length >ComConstants.Cmpp_MAX_MSG_LEN) || (bb.length
			// <ComConstants.Cmpp_MIN_MSG_LEN) ) {
			// PublicConstants.writeLog.info("Cmpp reSend pack length error :
			// "+bb.length,0 );
			// return;
			// }
			// if( PublicConstants.PackDEBUGLevel > 0 )
			// PublicConstants.writeLog.packDebug("Cmpp SEND to
			// :"+myid+"(bindType:"+myBindType+") \t",bb);

			if (sm.headCmdID == CmppConstants.Cmpp_Submit) {
				sm.firstSendtime = System.currentTimeMillis();
				sm.lastSendtime = System.currentTimeMillis();
				sm.sendCount = 1;
				vCmppSubmitMsg.add(sm);
			}
			synchronized (_lock) {
				myoutputstream.write(bb);
				myoutputstream.flush();
			}
		} catch (NullPointerException e) {
			LogManage.ins().myLog
					.severe(myBindType + "send pack is null: " + e);
		} catch (SocketException e) {
			LogManage.ins().myLog
					.severe(myBindType + "send disconnected: " + e);
			disconnect();
		} catch (IOException e) {
			LogManage.ins().myLog
					.severe(myBindType + "send disconnected: " + e);
			disconnect();
		}

	}

	public void send(byte[] bb) {
		try {
			if (bb == null) {
				// LogManage.ins().myLog .warning(myBindType+" send pack is null
				// ");
				return;
			}
			if ((bb.length > ComConstants.Cmpp_MAX_MSG_LEN)
					|| (bb.length < ComConstants.Cmpp_MIN_MSG_LEN)) {
				LogManage.ins().myLog.warning(myBindType
						+ " send pack length error:  " + bb.length);
				return;
			}
			synchronized (_lock) {
				myoutputstream.write(bb);
				myoutputstream.flush();
			}
		} catch (NullPointerException e) {

		} catch (SocketException e) {

			disconnect();
		} catch (IOException e) {

			disconnect();
		}
	}

	public void run() {
		while (!myShutdownFlag) {
			if (mystate != ComConstants.NOTCONNECTED) {
				try {
					byte[] buf = recv();
					if (buf.length == 1) {
						if (buf[0] == ComConstants.ERROR_INTERRUPT) {
							if ((System.currentTimeMillis() - mylastactivetime < ComConstants.Cmpp_MAX_NOTACTIVE_SECOND * 1000)
									// < 8*1000)
									&& (mystate.equals(ComConstants.WORKING))) {
								heartBeat();
							} else {
								disconnect();
								// LogManage.ins().myLog .warning(myBindType+"
								// sock timeout");
							}
						} else if (buf[0] == ComConstants.ERROR_SOCKET) {
							disconnect();
							// LogManage.ins().myLog .warning(myBindType+" sock
							// error ");
						}
					} else {
						mylastactivetime = System.currentTimeMillis();
						handle(buf);
					}
				} catch (Exception ee) {
					LogManage.ins().myLog.severe(myBindType
							+ " sock Exception " + ee);

				}
			} else {
				PublicFuction.threadSleep(2000);
			}
		} // end while
	}

	private byte[] recv() {
		byte[] receivedData;
		try {
			int dataSize = dataInStream.readInt();
			if ((dataSize < 16) || (dataSize > 1500)) {
				receivedData = new byte[1];
				receivedData[0] = ComConstants.ERROR_SOCKET;
				return receivedData;
			}
			receivedData = new byte[dataSize - 4];
			int dataTotalSizeToRead = dataSize - 4;
			int dataToReadLeft = dataSize - 4;
			int dataThisTimeRead;
			while (dataToReadLeft > 0) {
				dataThisTimeRead = dataInStream.read(receivedData,
						dataTotalSizeToRead - dataToReadLeft, dataToReadLeft);
				dataToReadLeft -= dataThisTimeRead;
			}
		} catch (NullPointerException e) {
			LogManage.ins().myLog.severe("  recv connection null: "
					+ e.toString());
			receivedData = new byte[1];
			receivedData[0] = ComConstants.ERROR_SOCKET;
		} catch (InterruptedIOException e) {
			receivedData = new byte[1];
			receivedData[0] = ComConstants.ERROR_INTERRUPT;
		} catch (Exception e) {
			LogManage.ins().myLog.severe("  recv socket: " + e.toString());
			receivedData = new byte[1];
			receivedData[0] = ComConstants.ERROR_SOCKET;
		}
		return receivedData;
	}

	public void handle(byte[] rcvdata) {
		int unPackCode = CmppConstants.error;
		int ret;
		boolean sendToSmsc = true;

		ShortMsgCmpp cmpp_sm = new ShortMsgCmpp();
		cmpp_sm.parsePackage(rcvdata);
		unPackCode = cmpp_sm.getParseResult();

		// debugMsg(cmpp_sm);

		switch (cmpp_sm.getCommandId()) {

		case CmppConstants.Cmpp_Submit_Resp:
			ComConstants.test_number++;
			LogManage.ins().myLog.logp(Level.INFO, "", "", "SUM_RSP:  "
					+ cmpp_sm.activeResult);
			if (cmpp_sm.activeResult != 0) {
				LogManage.ins().myLog.logp(Level.WARNING, "", "", "SUM_RSP:  "
						+ cmpp_sm.activeResult);

			}

			if (!addCmppMsg(cmpp_sm)) {
				LogManage.ins().myLog.logp(Level.WARNING, "", "",
						"add to Vecotor error:  " + cmpp_sm.activeResult);
			}
			break;

		case CmppConstants.Cmpp_Deliver:
			TestMOnum++;
			// if( mystate!=ComConstants.WORKING ) {
			// deliverRespSM.setHeadCmdStatus(CmppConstants.error_bind) ;
			// }
			// else if(unPackCode != CmppConstants.success) {
			// deliverRespSM.setHeadCmdStatus(unPackCode) ;
			// }
			// else if(! addCmppMsg(cmpp_sm) ) {
			// deliverRespSM.setHeadCmdStatus(CmppConstants.error) ;
			// }
			// else {
			// deliverRespSM.setHeadCmdStatus(CmppConstants.success) ;
			// }
			//
			// if( deliverRespSM.getHeadCmdStatus() != CmppConstants.success )
			// PublicConstants.writeLog.error("Cmpp Cmpp_Deliver from "+myid+"
			// error ! my.Retcode:"+cmpp_sm.getHeadCmdStatus()
			// +"\nheadSeqcNo:"+cmpp_sm.headSeqcNo
			// +"\nmsgid:"+cmpp_sm.msgId
			// +"\nmsgMode:"+cmpp_sm.msgMode
			// +"\nmsgSrcAddr:"+cmpp_sm.msgSrcAddr
			// +"\nmsgDestAddr:"+cmpp_sm.msgDestAddr
			// +"\nmsgDataCoding:"+cmpp_sm.msgDataCoding
			// +"\nmsgSmLength:"+cmpp_sm.msgLength
			// +"\nmsgContent:"+cmpp_sm.msgContent);
			addCmppMsg(cmpp_sm);
			deliverRespSM.setCommandId(CmppConstants.Cmpp_Deliver_Resp);
			deliverRespSM.headSeqcNo = cmpp_sm.headSeqcNo;
			deliverRespSM.submit_msg_id = cmpp_sm.submit_msg_id;
			send(deliverRespSM);
			break;
		case CmppConstants.Cmpp_Login_Resp:
			LogManage.ins().myLog.logp(Level.INFO, "", "", " Login rsp: "
					+ cmpp_sm.getHeadCmdStatus());
			// cmpp_sm.setConAuth(myMBEauth);
			// if( cmpp_sm.getRepAuthResult()==true ) {
			myLastVersion = cmpp_sm.getLastVersion();
			mystate = ComConstants.WORKING;

			break;
		case CmppConstants.Cmpp_Logout_Resp:
			mystate = ComConstants.NOTLOGIN;
			LogManage.ins().myLog
					.logp(Level.INFO, "", "", " Recv Logout  rsp ");
			break;
		case CmppConstants.Cmpp_Active:
			LogManage.ins().myLog.logp(Level.INFO, "", "", myBindType
					+ " Recv Active   rsp ");
			if (mystate != ComConstants.WORKING) {
				cmpp_sm.setHeadCmdStatus(CmppConstants.error_bind);
			} else {
				cmpp_sm.setHeadCmdStatus(CmppConstants.success);