/*
 * The contents of this file are subject to the terms
 * of the Common Development and Distribution License
 * (the License).  You may not use this file except in
 * compliance with the License.
 *
 * You can obtain a copy of the license at
 * https://glassfish.dev.java.net/public/CDDLv1.0.html or
 * glassfish/bootstrap/legal/CDDLv1.0.txt.
 * See the License for the specific language governing
 * permissions and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL
 * Header Notice in each file and include the License file
 * at glassfish/bootstrap/legal/CDDLv1.0.txt.
 * If applicable, add the following below the CDDL Header,
 * with the fields enclosed by brackets [] replaced by
 * you own identifying information
 * "Portions Copyrighted [year] [name of copyright owner]"
 *
 * Copyright (c) Ericsson AB, 2004-2007. All rights reserved.
 */

package com.ericsson.ssa.container;

import com.sun.grizzly.util.net.SSLImplementation;
import com.sun.grizzly.util.net.ServerSocketFactory;
import java.io.IOException;
import java.net.ConnectException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.ericsson.ssa.config.Config;
import com.ericsson.ssa.config.ConfigFactory;
import com.ericsson.ssa.config.Constants;
import com.ericsson.ssa.sip.Layer;
import com.ericsson.ssa.sip.LayerHelper;
import com.ericsson.ssa.sip.SipServletMessageImpl;
import com.ericsson.ssa.sip.SipServletRequestImpl;
import com.ericsson.ssa.sip.SipServletResponseImpl;
import com.ericsson.ssa.sip.dns.SipTransports;
import com.ericsson.ssa.sip.dns.TargetTuple;
import com.sun.grizzly.CallbackHandler;
import com.sun.grizzly.ConnectorHandler;
import com.sun.grizzly.Context;
import com.sun.grizzly.Context.KeyRegistrationState;
import com.sun.grizzly.Controller;
import com.sun.grizzly.DefaultPipeline;
import com.sun.grizzly.DefaultSelectionKeyHandler;
import com.sun.grizzly.IOEvent;
import com.sun.grizzly.Pipeline;
import com.sun.grizzly.ProtocolChain;
import com.sun.grizzly.DefaultProtocolChain;
import com.sun.grizzly.ProtocolChainInstanceHandler;
import com.sun.grizzly.TCPConnectorHandler;
import com.sun.grizzly.TCPSelectorHandler;
import com.sun.grizzly.UDPConnectorHandler;
import com.sun.grizzly.UDPSelectorHandler;
import com.sun.grizzly.Controller.Protocol;
import com.sun.grizzly.ProtocolFilter;
import com.sun.grizzly.SSLCallbackHandler;
import com.sun.grizzly.SSLConnectorHandler;
import com.sun.grizzly.filter.ReadFilter;
import com.sun.grizzly.filter.SSLReadFilter;
import com.sun.grizzly.SSLSelectorHandler;
import com.sun.grizzly.util.ByteBufferFactory;

import com.sun.grizzly.util.ByteBufferFactory.ByteBufferType;
import com.sun.grizzly.util.WorkerThread;
import java.io.UnsupportedEncodingException;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SocketChannel;
import javax.net.ssl.SSLContext;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * This class is responsible for initializing the Grizzly based SIP protocol
 * support in Sailfin.
 * 
 * @author ekrigro
 * @author Jeanfrancois Arcand
 */
public class GrizzlyNetworkManager extends NetworkManager implements Runnable {
    
    private final static String GRIZZLY_THREAD_COUNT = "sip.network.grizzly.threadcount";
    
    private final static String GRIZZLY_CLIENT_THREAD_POOL = "sip.network.grizzly.useServerThreadPool";
    
    private final static String GRIZZLY_CACHED_BYTEBUFFER = "sip.network.grizzly.maxCachedByteBuffer";
    
    public final static String SIP_CERTS = "Sip.Certs";
    
    private final Logger logger = Logger.getLogger("SipContainer");

    private Controller tcpController = new Controller();
    private Controller udpController = new Controller();
    private Controller tlsController = new Controller();
    
    private Controller tcpClientController = new Controller();
    private Controller udpClientController = new Controller();
    private Controller tlsClientController = new Controller();
    
    private Pipeline<Callable> ericssonThreadPoolWrapper = null;
    private Pipeline<Callable> clientPipeline = null;
    private Layer nextLayer = null;
    
    private static ConcurrentLinkedQueue<ByteBuffer> byteBuffersPool =
            new ConcurrentLinkedQueue<ByteBuffer>();
    
    private SSLContext sslContext;
    
    private Config _config = ConfigFactory.getConfig();
        
    private ConnectionManager connectionManager = null;
    
    private int threadPoolSize = 10;
    
    private final Lock writeLock = new ReentrantReadWriteLock().writeLock();
    
    public GrizzlyNetworkManager() {
        initPipeline();
    }
    
    public synchronized void start() throws IOException {
        try {
            int keepAliveTimeoutInSeconds = 600;
            
            InetAddress inet = InetAddress.getByName(_config.get(Constants.HOST));
            Integer port = Integer.parseInt( _config.get(Constants.SIP_PORT) );
            
            logger.info("Starting Grizzly SIP TCP Listener on port " + port);
            TCPSelectorHandler tcpSelector = createTcpEndpoint(
                    new InetSocketAddress(inet,port.intValue()) );
            tcpController.addSelectorHandler(tcpSelector);
            
            logger.info("Starting Grizzly SIP UDP Listener on port " + port);
            UDPSelectorHandler udpSelector = createUdpEndpoint(
                    new InetSocketAddress(inet,port.intValue()) );
            udpController.addSelectorHandler(udpSelector);
            
            logger.info("Starting Grizzly SIP TLS Listener on port " + (port+1));
            SSLSelectorHandler tlsSelector = createTLSEndpoint(
                    new InetSocketAddress(inet,(port.intValue()+1) ) );
            tlsController.addSelectorHandler(tlsSelector);
            
            DefaultSelectionKeyHandler keyHandler =
                    new DefaultSelectionKeyHandler(){
                public void expire(Iterator<SelectionKey> keys){
                } // The connections are never going to be closed
            };
            keyHandler.setLogger(logger);
            
            tcpSelector.setSelectionKeyHandler(keyHandler);
            udpSelector.setSelectionKeyHandler(keyHandler);
            tlsSelector.setSelectionKeyHandler(keyHandler);
            
            final DefaultProtocolChain protocolChain = new DefaultProtocolChain();
            protocolChain.setContinuousExecution(false);
            
            ReadFilter readFilter = new ReadFilter(){  
                
                private int bufferSize = 16 * 8192;
                
                public boolean execute(Context ctx) throws IOException {
                    SelectableChannel channel = ctx.getSelectionKey().channel();
               
                    // Make sure we remove that attribute in case the 
                    // ProtocolChain is not the default one.
                    ctx.removeAttribute(ProtocolFilter.SUCCESSFUL_READ);
                    
                    if (!channel.isOpen()){
                        ctx.setKeyRegistrationState
                                (KeyRegistrationState.CANCEL);
                        return false;
                    }
                    WorkerThread workerThread =
                            (WorkerThread)Thread.currentThread();    
           
                    // Re-size the ByteBuffer based on the underlying OS buffer.
                    if (workerThread.getByteBuffer().capacity() != bufferSize){
                         try{
                            if (ctx.getProtocol() == Protocol.TCP){
                                bufferSize = ((SocketChannel)channel)
                                    .socket().getReceiveBufferSize();
                            } else {
                                bufferSize = ((DatagramChannel)channel)
                                    .socket().getReceiveBufferSize();
                            }
                        } catch (IOException ex){
                            logger.log(Level.WARNING,"Bad Socket State: ",ex);
                            bufferSize = 16 * 8192;
                        }
                        workerThread.setByteBuffer(ByteBuffer.allocate(bufferSize));
                    }
                    return super.execute(ctx);
                }
            };
            readFilter.setContinuousExecution(false);
            
            protocolChain.addFilter(readFilter);
            protocolChain.addFilter(new MessageProcessorFilter(this));
            ProtocolChainInstanceHandler instanceHandler
                    = new ProtocolChainInstanceHandler(){
                public ProtocolChain poll(){
                    return protocolChain;
                }
                
                public boolean offer(ProtocolChain instance){
                    return true;
                }
            };
            // TCP
            tcpController.setProtocolChainInstanceHandler( instanceHandler );
            tcpController.setPipeline(ericssonThreadPoolWrapper);
            
            // UDP
            udpController.setProtocolChainInstanceHandler( instanceHandler );
            udpController.setPipeline(ericssonThreadPoolWrapper);
            
            // TLS
            tlsController.setProtocolChainInstanceHandler( instanceHandler );
            tlsController.setPipeline(ericssonThreadPoolWrapper);
            
            // TCP client
            tcpClientController.setProtocolChainInstanceHandler(instanceHandler);
            TCPSelectorHandler tcpClientSelectorHandler = new TCPSelectorHandler(true);
            tcpClientSelectorHandler.setSelectionKeyHandler(keyHandler);
            tcpClientController.addSelectorHandler(tcpClientSelectorHandler);
            tcpClientController.setPipeline(clientPipeline);
            
            // UDP Client
            udpClientController.setProtocolChainInstanceHandler(instanceHandler);
            UDPSelectorHandler udpClientSelectorHandler = new UDPSelectorHandler(true);
            udpClientSelectorHandler.setSelectionKeyHandler(keyHandler);
            udpClientController.addSelectorHandler(udpClientSelectorHandler);
            udpClientController.setPipeline(clientPipeline);

            // TLS support
            ServerSocketFactory serverSF = null;
            try{
                SSLImplementation sslHelper = SSLImplementation.getInstance();
                serverSF = sslHelper.getServerSocketFactory();
                serverSF.setAttribute("keystoreType","JKS");
                serverSF.setAttribute("keystore",
                        System.getProperty("javax.net.ssl.keyStore"));
                serverSF.setAttribute("truststoreType","JKS");
                serverSF.setAttribute("truststore",
                        System.getProperty("javax.net.ssl.trustStore"));
                //TODO Admin team to make it configurable
                String keyAlias = System.getProperty( 
                        "sip.network.tls.keyAlias", "s1as" );
                serverSF.setAttribute("keyAlias", keyAlias); //Default GF s1sa
                serverSF.init();
            } catch (ClassNotFoundException t){
                throw new RuntimeException(t);
            }
            sslContext = serverSF.getSSLContext();
            
            SSLReadFilter sslReadFilter = new SSLReadFilter(){
                public boolean execute(Context ctx) throws IOException {
                    SelectableChannel channel = ctx.getSelectionKey().channel();
               
                    // Make sure we remove that attribute in case the 
                    // ProtocolChain is not the default one.
                    ctx.removeAttribute(ProtocolFilter.SUCCESSFUL_READ);
                    
                    if (!channel.isOpen()){
                        ctx.setKeyRegistrationState
                                (KeyRegistrationState.CANCEL);
                        return false;
                    }
                  
                    boolean continueExecution = super.execute(ctx);
                    if (continueExecution && isNeedClientAuth()){
                        Object[] x509Cert = 
                                doPeerCertificateChain(ctx.getSelectionKey(),
                                                       isNeedClientAuth());
                        ctx.setAttribute(SIP_CERTS, x509Cert);
                    }
                    return continueExecution; 
                }  
            };
            sslReadFilter.setSSLContext(sslContext);
            //TODO Admin team to make it configurable
            Boolean clientAuth = new Boolean( 
                    System.getProperty( "sip.network.tls.clientAuth",
                        Boolean.FALSE.toString() ) );
            sslReadFilter.setNeedClientAuth(clientAuth);
            
            final ProtocolChain tlsProtocolChain = new DefaultProtocolChain();
            tlsProtocolChain.addFilter(sslReadFilter);
            tlsProtocolChain.addFilter(new MessageProcessorFilter(this));
            
            ProtocolChainInstanceHandler tlsProtocolChainHandler =
                    new ProtocolChainInstanceHandler(){
                public ProtocolChain poll(){
                    return tlsProtocolChain;
                }
                
                public boolean offer(ProtocolChain instance){
                    return true;
                }
            };
            tlsSelector.setProtocolChainInstanceHandler(tlsProtocolChainHandler);
            
            // TLS client
            tlsClientController.setProtocolChainInstanceHandler(tlsProtocolChainHandler);
            TCPSelectorHandler tlsClientSelectorHandler = new SSLSelectorHandler(true);
            tlsClientSelectorHandler.setSelectionKeyHandler(keyHandler);
            tlsClientController.addSelectorHandler(tlsClientSelectorHandler);
            tlsClientController.setPipeline(clientPipeline);            
                                    
            //Initiate the outbound connections pool
            connectionManager = new ConnectionManager();
                       
            ericssonThreadPoolWrapper.initPipeline();
            ericssonThreadPoolWrapper.startPipeline();
            
            clientPipeline.initPipeline();
            clientPipeline.startPipeline();                        
        } catch (IOException e) {
            logger.log(Level.WARNING,"Failed to start Grizzly",e);
            throw(e);
        }
    }
    
    public synchronized void stop() throws IOException {
        tcpController.stop();
        udpController.stop();
        tlsController.stop();
        tcpClientController.stop();
        udpClientController.stop();
        tlsClientController.stop();
    }
    
    void initPipeline() {
        
        threadPoolSize = Integer.parseInt(System.getProperty(GRIZZLY_THREAD_COUNT,String.valueOf(threadPoolSize)));
        
        Boolean useClientPipeline = Boolean.parseBoolean(System.getProperty(GRIZZLY_CLIENT_THREAD_POOL, "true"));
        
        Integer port = Integer.parseInt(_config.get(Constants.SIP_PORT));
        //TODO configurable thread pool size
        ericssonThreadPoolWrapper = new DefaultPipeline(threadPoolSize,threadPoolSize,"SipContainer",
                port);
        
        if (useClientPipeline){
            clientPipeline = new DefaultPipeline(threadPoolSize,threadPoolSize,"SipContainer-Client",
                    port);
        } else {
            clientPipeline = ericssonThreadPoolWrapper;
        }
        
        ((DefaultPipeline)ericssonThreadPoolWrapper)
                .setByteBufferType(ByteBufferType.HEAP);
        
        // No limits for now.
        ((DefaultPipeline)ericssonThreadPoolWrapper).setQueueSizeInBytes(-1);   
        clientPipeline.setQueueSizeInBytes(-1);  
        
        SipContainerThreadPool tp = SipContainerThreadPool.getInstance();
        tp.initialize(ericssonThreadPoolWrapper);
    }
    
    public void next(SipServletRequestImpl req) {
        LayerHelper.next(req, this, nextLayer);
    }
    
    public void next(SipServletResponseImpl resp) {
        LayerHelper.next(resp, this, nextLayer);
    }
    
    public void registerNext(Layer layer) {
        nextLayer = layer;
    }
    
    public void dispatch(SipServletRequestImpl req) {
        try {
            connectionManager.send(req);
        } catch (IOException e) {
            logger.log(Level.SEVERE,"Failed to send response : "+req.toDebugString(),e);
        }
        if( logger.isLoggable(Level.FINE)) {
            logger.fine("Done sending req : "+req.toDebugString());
        }
    }
    
    public void dispatch(SipServletResponseImpl resp) { //Failback, StreamResponseDispatchern is the normal path
        try {
            connectionManager.send(resp);
        } catch (IOException e) {
            logger.log(Level.SEVERE,"Failed to send response : "+resp.toDebugString(),e);
        }
        if( logger.isLoggable(Level.FINE)) {
            logger.fine("Done sending resp : "+resp.toDebugString());
        }
    }
    
    TCPSelectorHandler createTcpEndpoint(InetSocketAddress addr) {
        int keepAliveTimeoutInSeconds = 600;
        int linger = -1;
        
        TCPSelectorHandler selectorHandler = new TCPSelectorHandler();
        selectorHandler.setPort(addr.getPort());
        selectorHandler.setInet(addr.getAddress());
        selectorHandler.setLinger(linger);
        selectorHandler.setLogger(logger);
        selectorHandler.setReuseAddress(true);
        selectorHandler.setSocketTimeout(keepAliveTimeoutInSeconds * 1000);
        return selectorHandler;
    }
    
    SSLSelectorHandler createTLSEndpoint(InetSocketAddress addr) {
        int keepAliveTimeoutInSeconds = 600;
        int linger = -1;
        
        SSLSelectorHandler selectorHandler = new SSLSelectorHandler();
        selectorHandler.setPort(addr.getPort());
        selectorHandler.setInet(addr.getAddress());
        selectorHandler.setLinger(linger);
        selectorHandler.setLogger(logger);
        selectorHandler.setReuseAddress(true);
        selectorHandler.setSocketTimeout(keepAliveTimeoutInSeconds * 1000);
        return selectorHandler;
    }
    
    
    UDPSelectorHandler createUdpEndpoint(InetSocketAddress addr) {
        int keepAliveTimeoutInSeconds = 600;
        int linger = -1;
        
        UDPSelectorHandler selectorHandler = new UDPSelectorHandler();
        selectorHandler.setPort(addr.getPort());
        selectorHandler.setInet(addr.getAddress());
        selectorHandler.setLogger(logger);
        selectorHandler.setReuseAddress(true);
        return selectorHandler;
    }
    
    
    public void run() {
        try {
            // TODO: Use a thread pool
            new Thread(){
                public void run(){
                    try{
                        tcpClientController.start();
                    } catch (IOException ex){
                        logger.log(Level.SEVERE,"Exception, RUN ON GNM called!",ex);
                    }
                }
            }.start();
            new Thread(){
                public void run(){
                    try{
                        udpClientController.start();
                    } catch (IOException ex){
                        logger.log(Level.SEVERE,"Exception, RUN ON GNM called!",ex);
                    }
                }
            }.start();
            new Thread(){
                public void run(){
                    try{
                        udpController.start();
                    } catch (IOException ex){
                        logger.log(Level.SEVERE,"Exception, RUN ON GNM called!",ex);
                    }
                }
            }.start();
            new Thread(){
                public void run(){
                    try{
                        tlsController.start();
                    } catch (IOException ex){
                        logger.log(Level.SEVERE,"Exception, RUN ON GNM called!",ex);
                    }
                }
            }.start();
            new Thread(){
                public void run(){
                    try{
                        tlsClientController.start();
                    } catch (IOException ex){
                        logger.log(Level.SEVERE,"Exception, RUN ON GNM called!",ex);
                    }
                }
            }.start();    
            
            tcpController.start();
        } catch (IOException e) {
            logger.log(Level.SEVERE,"Exception, RUN ON GNM called!",e);
        }
    }


    /**
     * @return Returns the threadPoolSize.
     */
    public Integer getThreadPoolSize() {
       return new Integer(threadPoolSize);
    }



    /**
     * @param threadPoolSize
     *           The threadPoolSize to set.
     */
    public void setThreadPoolSize(Integer threadPoolSize) {
       if (threadPoolSize > 0)
          threadPoolSize = threadPoolSize;
    }

    
    class ConnectionManager {
        
        private int maxCachedBB = 50;
        
        private final static int DEFAULT_BB_SIZE = 32 * 8192;
        
        private ConcurrentHashMap<TargetTuple, ConnectorHandler> streams = new ConcurrentHashMap<TargetTuple, ConnectorHandler>();
        
        private int socketBufferSize = DEFAULT_BB_SIZE;
        
        private ConcurrentHashMap<TargetTuple,ConcurrentLinkedQueue<ByteBuffer>> asyncWriteQueue
                = new ConcurrentHashMap<TargetTuple,ConcurrentLinkedQueue<ByteBuffer>>();
        
        private ConcurrentHashMap<SelectableChannel,ByteBuffer> udpByteBuffers = 
                new ConcurrentHashMap<SelectableChannel,ByteBuffer>();
        
        
        public ConnectionManager() {
            maxCachedBB = Integer.parseInt(System.getProperty(GRIZZLY_CACHED_BYTEBUFFER,"50"));
        }
        
        
        private void send(SipServletMessageImpl sipServletMessage) 
            throws IOException {
            send(sipServletMessage,true,null);
        }
 
        
        private void send(SipServletMessageImpl sipServletMessage,
                          boolean useCache, ByteBuffer bb) throws IOException {            

            TargetTuple tt = sipServletMessage.getRemote();          
            ConnectorHandler handler = null;
            if (useCache){
                handler = streams.get(tt);
                
                if (handler != null && handler.getUnderlyingChannel() != null 
                        && !handler.getUnderlyingChannel().isOpen()){
                    try{
                        handler.close();
                    } finally {
                        streams.remove(tt); //Cleanup a broken connection
                    }
                    handler = null;
                }
            }

            int protocol = tt.getProtocol().ordinal();
            if(handler == null && protocol == SipTransports.TCP) {
                handler = createHandlerTCP(tt);
            } else if(handler == null && protocol == SipTransports.TLS) {
                handler = createHandlerTLS(tt);
            } else if(handler == null) {
                handler = createHandlerUDP(tt);
            }

            // The connection failed, returning null.
            if (handler == null) {
                throw new ConnectException("Connection refused to: " +
                        tt.getSocketAddress());
            }    
            
            boolean isConnected = true;
            if(protocol == SipTransports.TCP) {
                isConnected = ((TCPConnectorHandler)handler).isConnected();
            } else if(protocol == SipTransports.UDP) {
                isConnected = ((UDPConnectorHandler)handler).isConnected();
            } else if(protocol == SipTransports.TLS){
                isConnected = ((SSLConnectorHandler)handler).isConnected();
            }

            boolean blocking = (protocol == SipTransports.UDP ? false: true);
            if (bb == null){
                if (isConnected && socketBufferSize == DEFAULT_BB_SIZE){
                    try{
                        if (protocol == SipTransports.UDP){
                            socketBufferSize = ((DatagramChannel)handler
                                    .getUnderlyingChannel()).socket().
                                        getSendBufferSize();
                        } else {
                            socketBufferSize = ((SocketChannel)handler
                                    .getUnderlyingChannel()).socket().
                                        getSendBufferSize();
                        }
                    } catch (IOException ex){
                    }
                }  
                bb = toBuffer(sipServletMessage,socketBufferSize);
            }
                
            if (bb.position() == bb.limit() && bb.position() == 0){
                // Something when wrong
                logger.log(Level.SEVERE,"Invalid ByteBuffer");
                return;
            }
                 
            boolean cacheBuffer = true;
            try{
                try{
                    writeLock.lock();
                    if (isConnected){
                        ConcurrentLinkedQueue<ByteBuffer> bbs = asyncWriteQueue.remove(tt);
                        if (bbs != null){
                            try{
                                Iterator<ByteBuffer> i = bbs.iterator();
                                ByteBuffer bb2;
                                while(i.hasNext()){
                                    bb2 = i.next();
                                    handler.write(bb2,blocking);
                                    i.remove();
                                    bb2.clear();
                                    byteBuffersPool.offer(bb2);                               
                                }                                
                            } catch (IOException ex){
                                logger.log(Level.WARNING,"Write failed",ex);
                            }
                        }
                         
                        long nWrite = handler.write(bb, blocking);
                        if (nWrite == 0){
                            // UDP failed to writes the entire ByteBuffer.
                            udpByteBuffers.put(handler.getUnderlyingChannel(),bb);
                            cacheBuffer = false;
                        }
                    } else {
                        cacheBuffer = false;
                        boolean wasNull = false;
                        ConcurrentLinkedQueue<ByteBuffer> bbs = asyncWriteQueue.get(tt);
                        if (bbs == null){
                            bbs = new ConcurrentLinkedQueue<ByteBuffer>();
                            wasNull = true;
                        }
                       
                        // The queue is full. To avoid OOM, reject the ByteBuffer
                        // TODO: add a timer to clear the queue after a periods of X.                       
                        if (bbs.size() <= maxCachedBB){
                            bbs.offer(bb);
                            if (wasNull){
                                asyncWriteQueue.put(tt,bbs);
                            }
                        } else {
                            logger.log(Level.WARNING,"Unable to connect to:\n " 
                                    + tt + "\n Data will be lost.");
                        }
                    }
                    writeLock.unlock();
                } catch (IOException ex){      
                    if (logger.isLoggable(Level.FINE)){
                        logger.log(Level.FINE,"Unable to write",ex);
                    }
                    
                    try{
                        handler.close();
                    } finally {
                        streams.remove(tt); //Cleanup a broken connection
                    }

                    // Try with a non cached connection as the connection is 
                    // broken.
                    if (useCache) {
                        cacheBuffer = false;
                        send(sipServletMessage,false,bb);
                        return;
                    } else {
                        throw ex;
                    }
                }
            } finally {
                if (bb != null 
                        && bb.capacity() != DEFAULT_BB_SIZE && cacheBuffer){
                    bb.clear();
                    byteBuffersPool.offer(bb);   
                }
            }
        }
        
        
        /**
         * Return a pooled <code>ByteBuffer</code> containing the
         * SipMessage bytes.
         */
        private final ByteBuffer toBuffer(SipServletMessageImpl sipServletMessage,
                int size) throws UnsupportedEncodingException{
            ByteBuffer bb = byteBuffersPool.poll();
            if (bb == null){
                bb = ByteBuffer.allocate(size);
            }
            sipServletMessage.toBuffer(bb);
            bb.flip();
            return bb;
        }
        
        
        private synchronized ConnectorHandler createHandlerTCP(final TargetTuple tt) throws IOException {
            final InetSocketAddress remote = tt.getSocketAddress();
            final TCPConnectorHandler connectorHandler = (TCPConnectorHandler) 
                tcpClientController.acquireConnectorHandler(Protocol.TCP);    
            
            ConnectorHandler cached = streams.putIfAbsent(tt,connectorHandler);
            if(cached != null) {
                return cached;
            }
            
            CallbackHandler<Context> callbackHandler = new CallbackHandler<Context>(){
                public void onConnect(IOEvent<Context> ioEvent) {
                    SelectionKey key = ioEvent.attachment().getSelectionKey();
                    try{
                        if (!((SocketChannel)key.channel()).isConnectionPending()){
                            return;
                        }
                        connectorHandler.finishConnect(key);                                               
                        tcpClientController.registerKey(key, SelectionKey.OP_READ, 
                            Controller.Protocol.TCP);                   
                    } catch (IOException ex) {
                        if (logger.isLoggable(Level.FINE)){
                            logger.log(Level.FINE,"Connect failed" + remote,ex);
                        }
                        streams.remove(tt);
                        tcpClientController.getSelectorHandler(Protocol.TCP)
                            .getSelectionKeyHandler().cancel(key);                        
                    }
                }
                
                public void onRead(IOEvent<Context> ioEvent) {
                    try {
                        Context ctx = ioEvent.attachment();
                        SelectionKey key = ctx.getSelectionKey();
                        if (!key.isValid()){
                            streams.remove(tt);
                            return;
                        }
                        
                        // disable OP_READ on key before doing anything else
                        key.interestOps(key.interestOps() & (~SelectionKey.OP_READ));
                        ctx.getProtocolChain().execute(ioEvent.attachment());
                    } catch (Throwable e) {
                        logger.log(Level.SEVERE,"Read failed: ",e);
                    }
                }
                public void onWrite(IOEvent<Context> ioEvent) {
                    throw new IllegalStateException("Should not end up here!");
                }
            };
            try{
                connectorHandler.connect( remote, callbackHandler );
                return connectorHandler;
            } catch (Throwable t){
                if (logger.isLoggable(Level.FINE)){
                    logger.log(Level.FINE,"Connect failed" + remote,t);
                }
                return null;
            }
        }
        
        
        private final void waitOnLatch(CountDownLatch latch,
                int timeout, TimeUnit timeUnit) throws InterruptedException {
            latch.await(timeout, timeUnit);
        }
        
        
        private synchronized ConnectorHandler createHandlerTLS(final TargetTuple tt) throws IOException {
            final InetSocketAddress remote = tt.getSocketAddress();
            final SSLConnectorHandler connectorHandler = (SSLConnectorHandler)
                new SSLConnectorHandler(sslContext);
            
            ConnectorHandler cached = streams.putIfAbsent(tt,connectorHandler);
            if(cached != null) {
                return cached;
            }
            
            connectorHandler.setController(tlsClientController);           
            final ByteBuffer inBuffer = ByteBufferFactory.allocateView(false);            
            final CountDownLatch handshakeDoneLatch = new CountDownLatch(1);
            
            SSLCallbackHandler<Context> sslCallbackHander = new SSLCallbackHandler<Context>() {
                public void onConnect(IOEvent<Context> ioEvent) {
                    SelectionKey key = ioEvent.attachment().getSelectionKey();
                    try{
                        if (!((SocketChannel)key.channel()).isConnectionPending()){
                            return;
                        }
                        connectorHandler.finishConnect(key);
                        
                        key.attach(connectorHandler.getSSLEngine());

                        if (connectorHandler.handshake(inBuffer, true)) {
                            onHandshake(ioEvent);
                        }
                        connectorHandler.getSSLEngine().
                                getSession().putValue
                                (SSLReadFilter.HANDSHAKE,Boolean.TRUE);
                        tlsClientController.registerKey(key, 
                                SelectionKey.OP_READ, Controller.Protocol.TLS);
                    } catch (IOException ex) {
                        if (logger.isLoggable(Level.FINE)){
                            logger.log(Level.FINE,"Connect failed" + remote,ex);
                        }
                        streams.remove(tt);
                        tlsClientController.getSelectorHandler(Protocol.TLS)
                            .getSelectionKeyHandler().cancel(key);
                    }
                }
                
                // After handshake completes
                public void onHandshake(IOEvent<Context> ioEvent) {
                    inBuffer.clear();
                    handshakeDoneLatch.countDown();
                }
                
                // Data is ready to be read
                public void onRead(IOEvent<Context> ioEvent) {
                    try {
                        Context ctx = ioEvent.attachment();
                        SelectionKey key = ctx.getSelectionKey();
                        if (!key.isValid()){
                            streams.remove(tt);
                            key.cancel();
                            return;
                        }
                        
                        // disable OP_READ on key before doing anything else
                        key.interestOps(key.interestOps() & (~SelectionKey.OP_READ));
                        ctx.getProtocolChain().execute(ioEvent.attachment());
                    } catch (Throwable e) {
                        logger.log(Level.SEVERE,"Read failed: ",e);
                    }
                }
                
                // Channel is ready to write data
                public void onWrite(IOEvent<Context> ioEvent) {
                    throw new IllegalStateException("Should not end up here!");
                }
            };
            
            try{
                connectorHandler.connect( remote, sslCallbackHander );
                // This makes sure the handshake operations is completed
                // before marking this connectorHandler as ready. Might
                // be a performance bottleneck.
                waitOnLatch(handshakeDoneLatch,10, TimeUnit.SECONDS);
                return connectorHandler;
            } catch (Throwable t){
                if (logger.isLoggable(Level.FINE)){
                    logger.log(Level.FINE,"Connect failed" + remote,t);
                }
                return null;
            }
        }
        
        private synchronized ConnectorHandler createHandlerUDP(final TargetTuple tt) throws IOException {
            final InetSocketAddress remote = tt.getSocketAddress();
            final UDPConnectorHandler connectorHandler = (UDPConnectorHandler)
                udpClientController.acquireConnectorHandler(Protocol.UDP);
            
            ConnectorHandler cached = streams.putIfAbsent(tt,connectorHandler);
            if(cached != null) {
                return cached;
            }            
            
            CallbackHandler<Context> callbackHandler = new CallbackHandler<Context>(){
                public void onConnect(IOEvent<Context> ioEvent) {
                    SelectionKey key = ioEvent.attachment().getSelectionKey();
                    try{
                        connectorHandler.finishConnect(key);                         
                        if (connectorHandler.isConnected()) {
                            udpClientController.registerKey(key,SelectionKey.OP_READ,
                                    Controller.Protocol.UDP);
                        } else {
                            throw new IOException("Connect failed " + remote);
                        }
                    } catch(IOException ex) {
                        if (logger.isLoggable(Level.FINE)){
                            logger.log(Level.FINE,"Connect failed" + remote,ex);
                        }
                        streams.remove(tt);
                        udpClientController.getSelectorHandler(Protocol.UDP)
                            .getSelectionKeyHandler().cancel(key);                          
                    }
                }
                
                
                public void onRead(IOEvent<Context> ioEvent) {
                    try {
                        Context ctx = ioEvent.attachment();
                        SelectionKey key = ctx.getSelectionKey();                       
                        if (!key.isValid()){
                            streams.remove(tt);
                            return;
                        }
                        
                        // disable OP_READ on key before doing anything else
                        key.interestOps(key.interestOps() & (~SelectionKey.OP_READ));
                        ctx.getProtocolChain().execute(ioEvent.attachment());
                    } catch (Throwable e) {
                        logger.log(Level.SEVERE,"Read failed: ",e);
                    }
                }
                
                
                public void onWrite(IOEvent<Context> ioEvent) {
                    Context ctx = ioEvent.attachment();
                    SelectionKey key = ctx.getSelectionKey();   
                    DatagramChannel channel = (DatagramChannel)key.channel();
                    int nWrite = 0;
                    ByteBuffer bb = udpByteBuffers.remove(channel);
                    key.interestOps(key.interestOps() & (~SelectionKey.OP_WRITE));
                    try {
                        nWrite = channel.write(bb);
                        if (nWrite == 0){
                            key.attach(this);                    
                        }
                    } catch (IOException ex){
                        logger.log(Level.SEVERE,"Write failed: ",ex);
                    } finally {
                        if (nWrite !=0 && !bb.hasRemaining()){
                            byteBuffersPool.offer(bb);
                            
                        } else {
                            udpByteBuffers.put(channel,bb);
                            udpClientController.getSelectorHandler(Protocol.UDP)
                                .register(key, SelectionKey.OP_WRITE);    
                        }
                    }
                }
            };
            
            try{
                connectorHandler.connect( remote, callbackHandler );
                return connectorHandler;
            } catch (Throwable t){
                if (logger.isLoggable(Level.FINE)){
                    logger.log(Level.FINE,"Connect failed" + remote,t);
                }
                return null;
            }                
        }
        
    }
}
