import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import com.sun.grizzly.ConnectorHandler;
import com.sun.grizzly.Context;
import com.sun.grizzly.Controller;
import com.sun.grizzly.ControllerStateListenerAdapter;
import com.sun.grizzly.DefaultCallbackHandler;
import com.sun.grizzly.DefaultProtocolChain;
import com.sun.grizzly.ProtocolChain;
import com.sun.grizzly.ProtocolChainInstanceHandler;
import com.sun.grizzly.ProtocolFilter;
import com.sun.grizzly.UDPSelectorHandler;
import com.sun.grizzly.Controller.Protocol;
import com.sun.grizzly.connectioncache.client.CacheableConnectorHandlerPool;
import com.sun.grizzly.filter.ReadFilter;
import com.sun.grizzly.util.DefaultThreadPool;
import com.sun.grizzly.util.WorkerThread;


public class GrizzlyNullPointerException implements Runnable{
	
	private Controller serverController = new Controller();
	private Controller clientController = new Controller();
	private Thread thread;
	private final static int PORT = 5060;
	private final static int SEND_PORT = 5061;
	private final static String REMOTE_ADDRESS = "1.2.3.4";
	private final static int REMOTE_PORT = 5060;
	
	private class SendFilter implements ProtocolFilter{
		private InetSocketAddress local = null;

		public SendFilter(){
			try{
				InetAddress inetLocal = InetAddress.getLocalHost();
				this.local = new InetSocketAddress(inetLocal, SEND_PORT);
			}catch(Exception e){
				e.printStackTrace();
			}
		}
		
		public boolean execute(Context ctx) throws IOException {
	    	final WorkerThread workerThread = ((WorkerThread)Thread.currentThread());
	        ByteBuffer buffer = workerThread.getByteBuffer();
			buffer.flip();
			int limit = buffer.limit();
			byte[] msgBytes = new byte[limit];
			buffer.get(msgBytes, 0, limit);
			
			send(msgBytes, limit);
			return false;
		}
		
	    private void send(byte msg[], int length) throws IOException {
	   		ConnectorHandler handler = createHandlerUDP();
	        
	        ByteBuffer bb = toBuffer(msg, length);
	        boolean isConnected = handler.isConnected();
	        boolean cacheBuffer = true;
	        try {
	        	cacheBuffer = write(handler, bb, isConnected, msg, length);
	        } catch (IOException ex) {
	        	ex.printStackTrace();
	        	throw ex;
	        }finally{
	        	if (bb != null && cacheBuffer) {
	        		bb.clear();
	        	}
	            handler.close();
	            clientController.releaseConnectorHandler(handler);
	        }
	    }
	    
	    private ConnectorHandler createHandlerUDP() throws IOException {
	    	ConnectorHandler connectorHandler = clientController.acquireConnectorHandler(Protocol.UDP);
	    	DefaultCallbackHandler callbackHandler = new DefaultCallbackHandler(connectorHandler);
	    	
	    	try {
	    		connectorHandler.connect(new InetSocketAddress(REMOTE_ADDRESS, REMOTE_PORT), local, callbackHandler);
	    		return connectorHandler;
	    	} catch (IOException e) {
	    		throw e;
	    	}
	    }
	    
	   	private final ByteBuffer toBuffer(byte[] msg, int length) throws UnsupportedEncodingException {
	   		ByteBuffer bb = ((WorkerThread)Thread.currentThread()).getByteBuffer();
	   		bb.clear();
	        bb.put(msg, 0, length);
	        bb.flip();
	        return bb;
	    }
	   	
	    private boolean write(ConnectorHandler handler, ByteBuffer bb, boolean isConnected, byte[] msg, int length) throws IOException{
	    	long nWrite = 0L;
	    	if (isConnected)
	    		nWrite = handler.write(bb, false);
	    	if (nWrite == 0L || !isConnected){
				ByteBuffer copy = copy(msg, length);
				handler.writeToAsyncQueue(copy);
	    	}
	    	
	    	return isConnected;
	    }
	    
	    private ByteBuffer copy(byte[] msg, int length){
			ByteBuffer copy = ByteBuffer.allocate(length);
			copy.put(msg, 0, length);
			copy.flip();
			return copy;
	    }
	    
		public boolean postExecute(Context ctx) throws IOException {
			return true;
		}
	}
	
	private void doTest(){
		ProtocolChainInstanceHandler pciHandler = new ProtocolChainInstanceHandler() {
			final private ProtocolChain protocolChain = new DefaultProtocolChain();

        	public ProtocolChain poll() {
        		return protocolChain;
        	}

        	public boolean offer(ProtocolChain instance) {
        		return true;
        	}
		};
		
		ProtocolChain protocolChain = pciHandler.poll();
		ReadFilter readFilter = new ReadFilter();
		protocolChain.addFilter(readFilter);
		SendFilter sendFilter = new SendFilter();
		protocolChain.addFilter(sendFilter);
		
		DefaultThreadPool threadPool = new DefaultThreadPool();
		
		/* server */
		UDPSelectorHandler selectorHandler = new UDPSelectorHandler();
		selectorHandler.setPort(PORT);
		selectorHandler.setReuseAddress(true);
		serverController.addSelectorHandler(selectorHandler);
		serverController.setProtocolChainInstanceHandler(pciHandler);
		serverController.setThreadPool(threadPool);
		serverController.setAutoConfigure(false);
		serverController.setAllowContextCaching(true);
		
		/* client */
        clientController.setProtocolChainInstanceHandler(pciHandler);
		clientController.setThreadPool(threadPool);
		UDPSelectorHandler clientSelectorHandler = new UDPSelectorHandler(true);
        clientController.addSelectorHandler(clientSelectorHandler);
		clientController.setAutoConfigure(false);
		clientController.setAllowContextCaching(true);
		CacheableConnectorHandlerPool connectorHandlerPool = new CacheableConnectorHandlerPool(clientController, 100, 10, 1);
		clientController.setConnectorHandlerPool(connectorHandlerPool);
		clientController.setReadThreadsCount(8);
		
		this.thread = new Thread(this);
		thread.start();
	}
	
	public void run(){
        final CountDownLatch serverLatch = new CountDownLatch(1);
        serverController.addStateListener(new ControllerStateListenerAdapter() {
            @Override public void onReady() {
                serverLatch.countDown();
            }

            @Override public void onException(Throwable e) {
            	e.printStackTrace();
                if (serverLatch.getCount() > 0) {
                    serverLatch.countDown();
                }
            }
        });
        new Thread(serverController).start();
        
        try{
        	serverLatch.await(30, TimeUnit.SECONDS);
        }catch(InterruptedException e){
        	e.printStackTrace();
        }
        
        
        final CountDownLatch clientLatch = new CountDownLatch(1);
        clientController.addStateListener(new ControllerStateListenerAdapter() {
            @Override public void onReady() {
                clientLatch.countDown();
            }

            @Override public void onException(Throwable e) {
            	e.printStackTrace();
                if (clientLatch.getCount() > 0) {
                    clientLatch.countDown();
                }
            }
        });
        new Thread(clientController).start();
        
        try{
        	clientLatch.await(30, TimeUnit.SECONDS);
        }catch(InterruptedException e){
        	e.printStackTrace();
        }
	}
	
	public static void main(String[] args) {
		GrizzlyNullPointerException test = new GrizzlyNullPointerException();
		test.doTest();
	}

}