Small Proxy App based on Apache Mina 2.0.0-RC1

What I'm going to do is develop a small app which basically is going to broadcast a message received to another host in a defined port.
My small app is just a standalone app based on Apache Mina, is going to be listening on "X" port and it's going to broadcast message received to another app which is a blackbox. So simple, Isn't it?


What do you need?

• mina-filter-compression-2.0.0-RC1.jar
• mina-integration-xbean-2.0.0-RC1.jar
• slf4j-log4j12-1.5.0.jar
• mina-integration-beans-2.0.0-RC1.jar
• jzlib-1.0.7.jar
• jaxen-1.1.1.jar
• mina-transport-apr-2.0.0-RC1.jar
• log4j-1.2.15.jar
• slf4j-api-1.5.0.jar
• mina-core-2.0.0-RC1.jar

Or simple, you can get the full distribution and get them all in one single file.
This jar is just a luxury, but it's necesary, because I don't want to reinvent the wheel: commons-lang-2.3.jar

Well, time to work, the first thing you should code is the main class, my ProxyServer and 2 IoHandlers (ProxyHandlerAdapter & ClientHandlerAdapter).

And this is my code
public class GSBProxyServer { private static Logger logger = Logger.getLogger(GSBProxyServer.class); private int initPoolSize = Runtime.getRuntime().availableProcessors()+1; private int maxPoolSize = 20; private int keepAliveTimeInSecs = 10;//for each thread in our pool private static SocketAcceptor acceptor; private static ExecutorService executor; public GSBProxyServer(){ LogUtils.initialize(); if(logger.isInfoEnabled()){ logger.info("GSB TCP Proxy server starting."); } try { if(logger.isDebugEnabled()){ logger.debug("loding properties file."); } ReceiverFactory.load(Constants.PROPERTIES_FILE); } catch (Exception e) { e.printStackTrace(); logger.error("Tcp Receiver factory load failed."); System.exit(-1); } IoBuffer.setUseDirectBuffer(false);//switch the default buffer type to 'heap' IoBuffer.setAllocator(new SimpleBufferAllocator()); //IoBuffer.setAllocator(new CachedBufferAllocator()); executor = new OrderedThreadPoolExecutor(initPoolSize, maxPoolSize, keepAliveTimeInSecs, TimeUnit.SECONDS);//Ordered/UnorderedThreadPoolExecutor has prevention mechanism against OutOfMemoryError, so you will prefer these two classes to other Executor implementations probably. acceptor = new NioSocketAcceptor(initPoolSize); acceptor.getFilterChain().addLast("codec",new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8")))); acceptor.getFilterChain().addLast("threadPool", new ExecutorFilter(executor)); //acceptor.getSessionConfig().setReadBufferSize(2048);//by default is 2048 acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 1); acceptor.getSessionConfig().setTcpNoDelay(true); try { acceptor.setHandler(new ProxyHandlerAdapter()); acceptor.bind(new InetSocketAddress(Constants.PORT_NUMBER)); if(logger.isInfoEnabled()){ logger.info("GSB TCP Proxy server started. Listening on port " + Constants.PORT_NUMBER); } } catch (Exception e) { e.printStackTrace(); stop(); } } public static void main(String[] args) throws IOException { new GSBProxyServer(); } public static void stop() { logger.info("GSB TCP Proxy server stopping."); if (acceptor != null) { acceptor.unbind(); acceptor = null; } if (executor != null) { executor.shutdown(); executor = null; } } }

public class ProxyHandlerAdapter extends IoHandlerAdapter { private static Logger logger = Logger.getLogger(ProxyHandlerAdapter.class); private ClientHandlerAdapter clientHandlerAdapter; public ProxyHandlerAdapter(){ this.clientHandlerAdapter = new ClientHandlerAdapter(); } /** * When exception happens and it's caugth * * @param session * @param throwable */ public void exceptionCaught(IoSession session, Throwable throwable) throws Exception { throwable.printStackTrace(); session.close(false); } /** * Event for message received * * @param session * @param msg */ @SuppressWarnings("unchecked") public void messageReceived(IoSession session, Object msg) throws Exception { String str = msg.toString(); long initialTime = System.currentTimeMillis(); if(logger.isDebugEnabled()){ logger.debug("Entry messageReceived"); } try{ if(logger.isInfoEnabled()){ logger.info("messageReceived: " +str +" -from local:"+session.getLocalAddress()+" remote:"+session.getRemoteAddress()); } if (StringUtils.equalsIgnoreCase("quit", StringUtils.trim(str))) { session.close(true); if (clientHandlerAdapter != null) { clientHandlerAdapter.shutdown(); } return; } this.clientHandlerAdapter.broadcastMessage(str);//call clientHandler which is going to broadcast msg if(logger.isInfoEnabled()){ logger.info("send session acknowledge"); } session.write("ACK\r\n"); }catch (Exception ex) { logger.error(ex); ex.printStackTrace(); } if(logger.isDebugEnabled()){ logger.debug("Exit messageReceived: "+(System.currentTimeMillis()-initialTime)); } } }

public class ClientHandlerAdapter extends IoHandlerAdapter { private static Logger logger = Logger.getLogger(ClientHandlerAdapter.class); private int initPoolSize = Runtime.getRuntime().availableProcessors()+1; private int maxPoolSize = Constants.NUMBER_THREADS; private int keepAliveTimeInSecs = 5;//for each thread in our pool private int connectionTimeoutInMillis = 5 * 1000;//by default 5 seconds private SocketConnector socketConnector; private ExecutorService executor; public ClientHandlerAdapter(){ socketConnector = new NioSocketConnector(initPoolSize); executor = new OrderedThreadPoolExecutor(initPoolSize, maxPoolSize, keepAliveTimeInSecs, TimeUnit.SECONDS); socketConnector.getFilterChain().addLast("codec",new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8")))); socketConnector.getFilterChain().addLast("threadPool", new ExecutorFilter(executor));//One thread pool for most events socketConnector.getFilterChain().addLast("threadPool-msgSent", new ExecutorFilter(initPoolSize, maxPoolSize, keepAliveTimeInSecs, TimeUnit.SECONDS, IoEventType.MESSAGE_SENT)); socketConnector.getFilterChain().addLast("threadPool-msgReceived", new ExecutorFilter(initPoolSize, maxPoolSize, keepAliveTimeInSecs, TimeUnit.SECONDS, IoEventType.MESSAGE_RECEIVED)); socketConnector.setHandler(this); socketConnector.getSessionConfig().setTcpNoDelay(true);//No delays on TCP communication socketConnector.getSessionConfig().setKeepAlive(false); socketConnector.setConnectTimeoutMillis(connectionTimeoutInMillis);//set timeout } /** * When exception happens and it's caugth * * @param session * @param throwable */ public void exceptionCaught(IoSession session, Throwable t) throws Exception { t.printStackTrace(); try{ CloseFuture closeFuture = session.close(true); //closeFuture.awaitUninterruptibly();// Wait until the connection is closed? closeFuture.setClosed();// Now connection should be closed. }catch (Exception ignore) { //ignore } } /** * Broadcast message * * @param message */ @SuppressWarnings("unchecked") public void broadcastMessage(final String message) throws Exception { long initialTime = System.currentTimeMillis(); if(logger.isDebugEnabled()){ logger.debug("Entry broadcastMessage"); } try{ String appName = StringUtils.substring(message, 3, 6); if(logger.isDebugEnabled()){ logger.debug("appName:" + appName); } List<Receiver> aTcpReceiverList = ReceiverFactory.applicationMap.get(appName); if(aTcpReceiverList!=null){ for(Receiver aTcpReceiver: aTcpReceiverList){ try{ this.sendMessage(aTcpReceiver, message); }catch (Exception ex) { logger.error("Error trying to connect to "+aTcpReceiver.getHostName()+":"+aTcpReceiver.getPortNumber()+". "+ex.getMessage()); try{ this.sendMessage(aTcpReceiver, message); }catch (Exception newEx) { logger.error("Error trying to connect <<again>> to "+aTcpReceiver.getHostName()+":"+aTcpReceiver.getPortNumber()+". "+newEx.getMessage()); ex.printStackTrace(); } } } } }catch (Exception ex) { logger.error(ex); ex.printStackTrace(); } if(logger.isDebugEnabled()){ logger.debug("Exit broadcastMessage: "+(System.currentTimeMillis()-initialTime)); } } /** * Sends a given message to a Receiver * * @param aTcpReceiver * @param message * @throws Exception * * @see com.vzw.gsb.network.Receiver */ @SuppressWarnings("unused") private void sendMessageOld(final Receiver aTcpReceiver, final String message) throws Exception{ long initialTime = System.currentTimeMillis(); if(logger.isDebugEnabled()){ logger.debug("Entry sendMessage"); } IoSession session = null; ConnectFuture connectFuture = socketConnector.connect(new InetSocketAddress(aTcpReceiver.getHostName(),aTcpReceiver.getPortNumber())); connectFuture.awaitUninterruptibly();//Wait until the connection attempt is finished. session = connectFuture.getSession(); session.write(message);//write message to remote server if(logger.isInfoEnabled()){ logger.info("message sent to: "+aTcpReceiver.getHostName()+":"+aTcpReceiver.getPortNumber()); } if(logger.isDebugEnabled()){ logger.debug("Exit sendMessage: "+(System.currentTimeMillis()-initialTime)); } //connectFuture.cancel();//Cancels the connection attempt and notifies all threads waiting for this future? } /** * Sends a given message to a Receiver. * I used a IoFutureListener on my ConnectFuture to wait for acknowledgment that the client has connected to the server. * Once we know we are connected, we can start writing data to the server * * @param aTcpReceiver * @param message * @throws Exception * * @see com.vzw.gsb.network.Receiver * @see org.apache.mina.core.future.IoFutureListener * @see org.apache.mina.core.future.ConnectFuture */ private void sendMessage(final Receiver aTcpReceiver, final String message) throws Exception{ long initialTime = System.currentTimeMillis(); if(logger.isDebugEnabled()){ logger.debug("Entry sendMessage"); } ConnectFuture connectFuture = socketConnector.connect(new InetSocketAddress(aTcpReceiver.getHostName(),aTcpReceiver.getPortNumber())); connectFuture.addListener(new IoFutureListener<IoFuture>(){ public void operationComplete(IoFuture future) { /*long initialTime = System.currentTimeMillis(); if(logger.isInfoEnabled()){ logger.info("Entry operationComplete"); }*/ ConnectFuture connFuture = (ConnectFuture)future; if(connFuture.isConnected()){ IoSession session = future.getSession(); session.write(message);//write message to remote server if(logger.isInfoEnabled()){ logger.info("message sent to: "+aTcpReceiver.getHostName()+":"+aTcpReceiver.getPortNumber()); } }else{ logger.error("Timeout trying to connecto to: "+aTcpReceiver.getHostName()+":"+aTcpReceiver.getPortNumber()); } /*if(logger.isInfoEnabled()){ logger.info("Exit operationComplete: "+(System.currentTimeMillis()-initialTime)); }*/ } }); if(logger.isDebugEnabled()){ logger.debug("Exit sendMessage: "+(System.currentTimeMillis()-initialTime)); } } /** * Event when message is received (ACK) from target server * * @param session * @param message */ @Override public void messageReceived(IoSession session, Object message) throws Exception { long initialTime = System.currentTimeMillis(); if(logger.isInfoEnabled() && StringUtils.isNotEmpty(message.toString())){//ACK coming from remote server logger.info("messageReceived: " +message);//+" -from local:"+session.getLocalAddress()+" remote:"+session.getRemoteAddress()+", closing session & closeFuture"); } if(logger.isDebugEnabled() && StringUtils.isNotEmpty(message.toString())){ logger.debug("Entry messageReceived"); } try{//once ACK comes, I close this session CloseFuture closeFuture = session.close(true); //closeFuture.awaitUninterruptibly();// Wait until the connection is closed? closeFuture.setClosed();// Now connection should be closed. if(logger.isDebugEnabled() && StringUtils.isNotEmpty(message.toString())){ logger.debug("isFutureClosed?"+closeFuture.isClosed()+", isSessionClosing?"+session.isClosing()); } }catch (Exception ignore) { //ignore } if(logger.isDebugEnabled() && StringUtils.isNotEmpty(message.toString())){ logger.debug("Exit messageReceived: "+(System.currentTimeMillis()-initialTime)); } } /** * Release all resources (SocketConnector) and ExecutorService * * @throws Exception */ public void shutdown() throws Exception { if(socketConnector!=null){ socketConnector.dispose(); } if (executor != null) { executor.shutdown(); } } }

How can I test it??? Just by create a client using java.net.Socket or just by using Telnet to that particular port

I didn't include other classes because they're not important