serverhandler.java

短信开发用于文件交换处理转发的类模块

package com.pub.backserver.net;

import org.apache.log4j.Logger;
import org.apache.mina.common.ByteBuffer;
import org.apache.mina.common.IdleStatus;
import org.apache.mina.common.IoHandlerAdapter;
import org.apache.mina.common.IoSession;
import org.apache.mina.common.TransportType;
import org.apache.mina.common.WriteFuture;
import org.apache.mina.filter.SSLFilter;
import org.apache.mina.transport.socket.nio.SocketSessionConfig;

import com.pub.backserver.EntityHead;
import com.pub.backserver.IEntity;
import com.pub.backserver.blackList.BlackEntity;
import com.pub.backserver.order.OrderEntity;
import com.pub.backserver.phase.PhaseEntity;
import com.pub.util.ByteBufferOverplus;

public class ServerHandler extends IoHandlerAdapter {
	private static Logger log = Logger.getLogger(ServerHandler.class);

	private ServerHandlerParameter chp = null;

	private ICallback callback;

	public ServerHandler(ICallback callback, ServerHandlerParameter chp) {
		this.callback = callback;
		this.chp = chp;
	}

	public void sessionCreated(IoSession session) {
		session.setAttribute("IsOverplus", false);
		ByteBufferOverplus bo = new ByteBufferOverplus(chp.inBufferSize / 2);
		session.setAttribute("Overplus", bo);
		session.setIdleTime(IdleStatus.BOTH_IDLE, chp.IdleTime);
		if (session.getTransportType() == TransportType.SOCKET) {
			((SocketSessionConfig) session.getConfig())
					.setReceiveBufferSize(chp.inBufferSize);
			((SocketSessionConfig) session.getConfig())
					.setSendBufferSize(chp.outBufferSize);
		}
		// We're going to use SSL negotiation notification.
		session.setAttribute(SSLFilter.USE_NOTIFICATION);
	}

	public void sessionIdle(IoSession session, IdleStatus status) {
		ServerHandlerParameter chp = (ServerHandlerParameter) session
				.getAttribute("Parameter");
		// 关闭连接，清理连接
		session.close();
	}

	public void exceptionCaught(IoSession session, Throwable cause) {
		log.error("", cause);
		// Close connection when unexpected exception is caught.
		session.close();
	}

	public void messageReceived(IoSession session, Object message)
			throws Exception {
		if (!(message instanceof ByteBuffer)) {
			return;
		}
		// Server_Handler_Parameter chp = (Server_Handler_Parameter) session.
		// getAttribute("Parameter");
		boolean isOverplus = (Boolean) session.getAttribute("IsOverplus");
		ByteBufferOverplus overplus = (ByteBufferOverplus) session
				.getAttribute("Overplus");
		// ByteBuffer msg = (ByteBuffer) message;
		ByteBuffer inBuf = null;

		boolean switchover = true;

		int len = 0;
		int cmd = 0;
		try {
			while (true) {
				len = 0;
				cmd = 0;

				if (isOverplus) {

					session.setAttribute("IsOverplus", false);
					// 测试一下
					isOverplus = (Boolean) session.getAttribute("IsOverplus");
					isOverplus = false;
					// 填充结尾
					overplus.setBufferTail((ByteBuffer) message);
					// 获得上次遗留数据
					inBuf = overplus.getBuffer();
					log.debug("isOverplus : " + inBuf);
				} else {

					// 转换处理的buf
					if (switchover) {
						inBuf = (ByteBuffer) message;
						switchover = false;
					}

					// 判断接收长度是否可以读取头长度
					if (inBuf.limit() - inBuf.position() == 0) {
						break;
					} else if (inBuf.limit() - inBuf.position() < 4) {
						session.setAttribute("IsOverplus", true);
						// 填充头部
						overplus.setBufferHead(inBuf);
						break;
					}

					// 判断接收长度是否等于处理长度
					len = inBuf.getInt(inBuf.position());
					if (inBuf.limit() - inBuf.position() < len) {
						session.setAttribute("IsOverplus", true);
						// 填充头部
						overplus.setBufferHead(inBuf);
						break;
					}

				}

				len = inBuf.getInt(inBuf.position());
				cmd = inBuf.getInt(inBuf.position() + 4);
				log.debug("clientID : " + session.getAttribute("clientID")
						+ "  len : " + len + "  cmd : "
						+ Integer.toHexString(cmd) + "  " + inBuf.buf());
				Integer clientID = (Integer) session.getAttribute("clientID");

				switch (cmd) {
				case EntityHead.QUERY_BLACK: {
					BlackEntity be = new BlackEntity();
					be.readPackage(inBuf);
					int ret = 0;
					ret = callback.queryBlack(be);
					if (ret < 0) {
						if (ret != -3) {
							be.setType(-1); // error
						} else {
							be.setType(ret); // error
						}
					}
					be.setCommandID(be.QUERY_BLACK_RESP);
					sessionWrite(session, be, false);
					break;
				}
				case EntityHead.INSERT_BLACK: {
					BlackEntity be = new BlackEntity();
					be.readPackage(inBuf);
					int ret = callback.insertBlack(be);
					be.setTotalLength(ret);
					be.setCommandID(be.INSERT_BLACK_RESP);
					sessionWrite(session, be, false);
					break;
				}
				case EntityHead.DELETE_BLACK: {
					BlackEntity be = new BlackEntity();
					be.readPackage(inBuf);
					int ret = callback.deleteBlack(be);
					be.setTotalLength(ret);
					be.setCommandID(be.DELETE_BLACK_RESP);
					sessionWrite(session, be, false);
					break;
				}
				case EntityHead.QUERY_PHASE: {
					PhaseEntity pe = new PhaseEntity();
					pe.readPackage(inBuf);
					int ret = 0;
					ret = callback.queryPhase(pe);
					if (ret != 0) {
						pe.setProvID((short) ret);
						pe.setCityID((short) ret);
						pe.setGwID((short) ret);
					}

					pe.setCommandID(pe.QUERY_PHASE_RESP);
					sessionWrite(session, pe, false);
					break;

				}
				case EntityHead.INSERT_PHASE: {
					PhaseEntity pe = new PhaseEntity();
					pe.readPackage(inBuf);
					int ret = callback.insertPhase(pe);
					pe.setTotalLength(ret);
					pe.setCommandID(pe.INSERT_PHASE_RESP);
					sessionWrite(session, pe, false);
					break;
				}
				case EntityHead.DELETE_PHASE: {
					PhaseEntity pe = new PhaseEntity();
					pe.readPackage(inBuf);
					int ret = callback.deletePhase(pe);
					pe.setTotalLength(ret);
					pe.setCommandID(pe.DELETE_PHASE_RESP);
					sessionWrite(session, pe, false);
					break;
				}
				case EntityHead.QUERY_ORDER: {
					OrderEntity oe = new OrderEntity();
					oe.readPackage(inBuf);
					int ret = 0;
					ret = callback.queryOrder(oe);
					if (ret != 0) {
						oe.setOperation((short) ret);
						oe.setProvinceId((short) ret);
						oe.setGwId(ret);
						oe.setChannelId(ret);
					}
					oe.setCommandID(oe.QUERY_ORDER_RESP);
					sessionWrite(session, oe, false);
					break;

				}
				case EntityHead.INSERT_ORDER: {
					OrderEntity oe = new OrderEntity();
					oe.readPackage(inBuf);
					int ret = callback.insertOrder(oe);
					oe.setTotalLength(ret);
					oe.setCommandID(oe.INSERT_ORDER_RESP);
					sessionWrite(session, oe, false);
					break;
				}
				case EntityHead.DELETE_ORDER: {
					OrderEntity oe = new OrderEntity();
					oe.readPackage(inBuf);
					int ret = callback.deleteOrder(oe);
					oe.setTotalLength(ret);
					oe.setCommandID(oe.DELETE_ORDER_RESP);
					sessionWrite(session, oe, false);
					break;
				}
				default:
					log.debug("Expanded buf : " + inBuf.buf() + "  len : "
							+ len + "  cmd : " + Integer.toHexString(cmd));
					throw new Exception("Expanded  buf : " + inBuf.buf()
							+ "  len : " + len + "  cmd : "
							+ Integer.toHexString(cmd));
				}
				// log.debug("after buf : " + inBuf.buf() + " len : " + len +
				// " cmd : " + Integer.toHexString(cmd));

			}
		} finally {
			try {
				// log.debug("after buf : " + inBuf.buf());
				// inBuf.release();
			} catch (Throwable ex) {
				log.error("", ex);
			}
		}
	}

	private boolean sessionWrite(IoSession session, IEntity ch, boolean isJoin)
			throws Exception {
		boolean flag = false;
		ByteBuffer outBuf = null;
		outBuf = ByteBuffer.allocate(128).setAutoExpand(true);
		ch.writePackage(outBuf);
		outBuf.flip();
		WriteFuture wf = session.write(outBuf);
		if (isJoin) {
			wf.join();
			flag = wf.isWritten();
		} else {
			flag = true;
		}
		log.debug(ch);
		return flag;
	}

	// public ServerHandler() {
	// }
	//
	// public void onClose() {
	// }
	//
	// public void onReload(Object reloadparam) {
	// }
	//
	// public String onStat() {
	// return "";
	// }
	//
	// public int getHeaderLength() {
	// return 8; //len(int)|cmd(int)
	// }
	//
	// public int getPackageLength(ByteBuffer inBuf) {
	// return inBuf.getInt(0);
	// }
	//
	// private IServer server;
	//
	// public int ProcessPackage(ByteBuffer inBuf) {
	// int len = inBuf.getInt();
	// int cmd = inBuf.getInt();
	// // log.debug(Integer.toHexString(cmd));
	// outBuf.compact();
	// int pos = outBuf.position();
	// outBuf.putInt(0); //len
	// if (cmd == NetCmds.QUERY_BLACK) {
	// outBuf.putInt(NetCmds.QUERY_BLACK_RESP); //cmd
	// BlackEntity be = new BlackEntity();
	// be.fromByteBuffer(inBuf);
	// int ret = server.queryBlack(be);
	// if (ret < 0) {
	// if (ret != -3) {
	// be.setType( -1); //error
	// } else {
	// be.setType(ret); //error
	// }
	// }
	// be.toByteBuffer(outBuf);
	// } else if (cmd == NetCmds.QUERY_CHANNEL) {
	// outBuf.putInt(NetCmds.QUERY_CHANNEL_RESP); //cmd
	// ChannelEntity ce = new ChannelEntity();
	// ce.fromByteBuffer(inBuf);
	// ce.set_numchannelid((short) 0);
	// server.queryChannel(ce);
	// ce.toByteBuffer(outBuf);
	// } else if (cmd == NetCmds.QUERY_OUTNUMBER) {
	// outBuf.putInt(NetCmds.QUERY_OUTNUMBER_RESP); //cmd
	// OutnumberEntity oe = new OutnumberEntity();
	// oe.fromByteBuffer(inBuf);
	// int ret = server.queryOutNumber(oe);
	// // log.debug(oe);
	// oe.toByteBuffer(outBuf);
	// } else if (cmd == NetCmds.QUERY_PHASE) {
	// outBuf.putInt(NetCmds.QUERY_PHASE_RESP); //cmd
	// PhaseEntity pe = new PhaseEntity();
	// pe.fromByteBuffer(inBuf);
	// server.queryPhase(pe);
	// pe.toByteBuffer(outBuf);
	// } else if (cmd == NetCmds.INSERT_BLACK) {
	// outBuf.putInt(NetCmds.INSERT_BLACK_RESP); //cmd
	// BlackEntity be = new BlackEntity();
	// be.fromByteBuffer(inBuf);
	// int ret = server.saveBlack(be);
	// be.toByteBuffer(outBuf);
	// } else if (cmd == NetCmds.INSERT_CHANNEL) {
	// outBuf.putInt(NetCmds.INSERT_CHANNEL_RESP); //cmd
	// ChannelEntity ce = new ChannelEntity();
	// ce.fromByteBuffer(inBuf);
	// int ret = server.saveChannel(ce);
	// //ce.set_numchannelid((short) ret);
	// ce.toByteBuffer(outBuf);
	// } else if (cmd == NetCmds.INSERT_OUTNUMBER) {
	// outBuf.putInt(NetCmds.INSERT_OUTNUMBER_RESP); //cmd
	// OutnumberEntity oe = new OutnumberEntity();
	// oe.fromByteBuffer(inBuf);
	// if (oe.getMobile().equals("") ||
	// oe.getServiceID() == 0 ) {
	// int ret = server.queryOutNumber(oe);
	// } else {
	// int ret = server.saveOutNumber(oe);
	// }
	// oe.toByteBuffer(outBuf);
	// } else if (cmd == NetCmds.INSERT_SVROUTNUMBER) {
	// outBuf.putInt(NetCmds.INSERT_SVROUTNUMBER_RESP); //cmd
	// OutnumberEntity oe = new OutnumberEntity();
	// oe.fromByteBuffer(inBuf);
	// if (oe.getMobile().equals("") ||
	// oe.getServiceID() == 0 ) {
	// int ret = server.queryOutNumber(oe);
	// } else {
	// int ret = server.saveSvrOutNumber(oe);
	// }
	// oe.toByteBuffer(outBuf);
	// } else if (cmd == NetCmds.INSERT_PHASE) {
	// outBuf.putInt(NetCmds.INSERT_PHASE_RESP); //cmd
	// PhaseEntity pe = new PhaseEntity();
	// pe.fromByteBuffer(inBuf);
	// int ret = server.savePhase(pe);
	// pe.toByteBuffer(outBuf);
	// }
	//
	// len = outBuf.position() - pos;
	// outBuf.putInt(pos, len);
	//
	// outBuf.flip();
	//
	// //try{this.Send(null);}catch(Throwable ex){}
	//
	// return 1;
	// }
	//
	// public int getData(ByteBuffer outBuf) {
	// return 0;
	// }
	//
	// public void onShutdown(boolean reconnect) {
	// }
}