cmppserver.java.svn-base

华为短信网关实现的功能,包括服务器端和客户端代码

package com.fetion.cmpp.server;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.serialization.ObjectSerializationCodecFactory;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;

import com.fetion.cmpp.common.Message;
import com.fetion.cmpp.common.util.ConfigException;
import com.fetion.cmpp.server.conf.ServerConfig;
import com.fetion.cmpp.server.db.DBTool;
import com.fetion.cmpp.server.db.DBToolImp;
import com.fetion.cmpp.server.sender.CmppSenderServer;
import com.fetion.cmpp.server.util.ConfigUtil;
import com.fetion.cmpp.server.util.ServerException;
import com.sleepycat.je.DatabaseException;

/**
 * 服务器端核心程序
 * 包括：信息接收模块和信息发送模块
 * @author Administrator
 *
 */
public class CmppServer implements ServerCallBack, CmppServerMBean {

	private final Log logger = LogFactory.getLog(getClass());

	private int port;

	private NioSocketAcceptor acceptor;

	private DBTool dbTool = new DBToolImp();

	private CmppSenderServer senderServer;

	private boolean active = false;

	public CmppServer() {
		ServerConfig config;
		try {
			config = ConfigUtil.loadConfig();
			port = config.getCmppServerPort();
			senderServer = new CmppSenderServer(config.getShedule_Expression());
		} catch (ConfigException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	public boolean isActive(){
		return active;
	}

	public void start() {
		if (active)
			logger.info("server had started !");
		else {
			logger.info("server starting ..... ");

			// run receive server
			acceptor = new NioSocketAcceptor();
			// Prepare the service configuration.
			acceptor.getFilterChain().addLast(
					"codec",
					new ProtocolCodecFilter(
							new ObjectSerializationCodecFactory()));
			acceptor.getFilterChain().addLast("logger", new LoggingFilter());

			acceptor.setHandler(new ServerSessionHandler(this));
			try {
				acceptor.bind(new InetSocketAddress(port));
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

			// run sender server
			try {
				if (senderServer != null) {
					senderServer.run();
				}
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			active = true;
			logger.info("server start on port " + port);
		}
	}

	public void stop() {
		if (active) {
			logger.info("server stoping ...... ");
			if (acceptor != null) {
				acceptor.dispose();
				acceptor = null;
			}
			if (senderServer != null) {
				senderServer.stop();
			}
			active = false;
			logger.info("server stoped");
		} else
			logger.info("server had stoped!");
	}

	public void stopWithExit() {
		if (active) {
			logger.info("server stoping ...... ");
			if (acceptor != null) {
				acceptor.dispose();
				acceptor = null;
			}
			if (senderServer != null) {
				senderServer.stop();
			}
			System.exit(0);
			active = false;
			logger.info("server exited");
		} else
			logger.info("server had stoped!");
	}

	public void reStart() {
		stop();
		start();
	}

	public void onReceived(Message message) throws ServerException{
		if (dbTool != null) {
			try {
				dbTool.openDB();
				dbTool.insertNewMessages(message);
				dbTool.closeDB();
				logger.info("new Message added : " + message.toString());
			} catch (Exception e) {
				throw new ServerException("save new message error !",e);
			}
		}
	}
	
	public int getAllNewMessages() {
		List<Message> news = new ArrayList<Message>();
		try {
			dbTool.openDB();
			Map<Long, Message> newMessages = dbTool.getAllNewMessages();
			if (newMessages != null && newMessages.size()>0) {
				news.addAll(newMessages.values());
				logger.info("未读的信息有" + news.size() + "条");
				for (Message m : news) {
					logger.debug("未读的信息有： " + m);
				}
			}else {
				logger.info("没有未读的信息");
			}
			dbTool.closeDB();
		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (DatabaseException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return news.size();
	}

	public int getAllHistoryMessages() {
		List<Message> historys = new ArrayList<Message>();
		try {
			dbTool.openDB();
			Map<Long, Message> historyMessages = dbTool.getAllHistoryMessages();
			if (historyMessages != null && historyMessages.size()>0) {
				historys.addAll(historyMessages.values());
				logger.info("历史信息有" + historys.size() + "条");
				for (Message m : historys) {
					logger.debug("历史信息有： " + m);
				}
			} else {
				logger.debug("没有历史信息");
			}
			dbTool.closeDB();
		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (DatabaseException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return historys.size();
	}


	public void deleteAllHistoryMessages() {
		try {
			dbTool.openDB();
			dbTool.deleteAllHistoryMessages();
			dbTool.closeDB();
		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (DatabaseException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	public void deleteAllNewMessages() {
		try {
			dbTool.openDB();
			dbTool.deleteAllNewMessages();
			dbTool.closeDB();
		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (DatabaseException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public void deleteDataBase() {
		try {
			dbTool.deleteDataBase();
		} catch (DatabaseException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}		
	}
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		CmppServer server = new CmppServer();
		server.start();
		// try {
		// Thread.sleep(10000);
		// } catch (InterruptedException e) {
		// // TODO Auto-generated catch block
		// e.printStackTrace();
		// }
	}

}