/*
 * 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.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
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 com.ericsson.ssa.config.annotations.Configuration;
import com.ericsson.ssa.config.annotations.UpdatePolicy;
import com.ericsson.ssa.config.annotations.UsagePolicy;
import com.ericsson.ssa.container.startup.SipMonitoring;
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.TargetResolver;
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.net.Socket;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Map;
import javax.net.ssl.SSLContext;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.jvnet.glassfish.comms.util.LogUtil;

/**
 * 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 IS_CLIENT_EXECUTION = "isClientExecution";
    public final static int DEFAULT_BB_SIZE = 32 * 8192;
    private static final String SERVER_PIPELINE_NAME = "SipContainer-servers"; //Only used for name of Pipelines.
    private static final String CLIENT_PIPELINE_NAME = "SipContainer-clients"; //Only used for name of Pipelines.
    private static final int DEFAULT_PIPELINE_DISTINCTION_PORT = 5060; //Only used for unique ID of the default Pipeline.
    public final static String SIP_CERTS = "Sip.Certs";
    private final LogUtil logger = LogUtil.SIP_LOGGER;
    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> serverPipeline = null;
    private Pipeline<Callable> clientPipeline = null;
    private Layer nextLayer = null;
    static ConcurrentLinkedQueue<ByteBuffer> byteBuffersPool =
            new ConcurrentLinkedQueue<ByteBuffer>();
    private SSLContext sslContext;
    private ConnectionManager connectionManager = null;
    private final Lock writeLock = new ReentrantReadWriteLock().writeLock();
    //properties
    private int keepAliveTimeoutInSeconds = 600;
    private String keyStore = null;
    private String trustStore = null;
    private int maxPipelineThreads = 10; //100
    private int minPipelineThreads = 10; //20
    private int pipelineThreadsIncrement = 1;
    private int pipelineInitialByteBufferSize = 8192; //same as DefaultPipeline
    private boolean clientsShareServerThreadPool = true;
    private int maxCachedByteBuffers = 50;
    private Map<String, SipBindingCtx> ctxs =
            new HashMap<String, SipBindingCtx>();

    public GrizzlyNetworkManager() {
        initPipeline();
    }

    public synchronized void start() {
        final DefaultSelectionKeyHandler keyHandler =
                new DefaultSelectionKeyHandler() {

                    @Override
                    public void expire(Iterator<SelectionKey> keys) {
                    } // The connections are never going to be closed
                    
                    @Override
                    public void cancel(SelectionKey key) {
                        super.cancel(key);
                        connectionManager.remove(key);
                    }                     
                };
        keyHandler.setLogger(logger.getLogger());

        for (String ctx : SipBindingResolver.instance().getLocalContexts()) {
            startCtxServers(ctx, keyHandler);
        }

        SipBindingResolver.instance().registerSipBindingListener(new SipBindingListener() {

            public void newSipBindingCtxAvaliable(String context) {
                startCtxServers(context, keyHandler);
            }

            public void sipBindingCtxUpdated(String context) {
                restartCtxServers(context, keyHandler);
            }

            public void sipBindingCtxRemoved(String context) {
                stopCtxServers(context, keyHandler);
            }
            });

        //Initiate the outbound connections pool
        connectionManager = new ConnectionManager();

        serverPipeline.initPipeline();
        serverPipeline.startPipeline();

        clientPipeline.initPipeline();
        clientPipeline.startPipeline();
    }

    public void startCtxServers(String ctx, DefaultSelectionKeyHandler keyHandler) {
        SipBindingCtx sipBindingCtx = SipBindingResolver.instance().getContext(ctx);

        if (sipBindingCtx != null && !sipBindingCtx.isStale()) {
            ctxs.put(ctx, sipBindingCtx);
            final ProtocolChain udpProtocolChain = getProtocolChain();
            ProtocolChainInstanceHandler instanceHandler =
                    new SimpleProtocolChainInstanceHandler(udpProtocolChain);

            for (TargetTuple targetTuple : sipBindingCtx.getTargetTuples()) {
                if (logger.logInfo()) {
                    logger.info("sip.stack.network.starting_sip_binding", "\n" + targetTuple);
                }
                switch (targetTuple.getProtocol().ordinal()) {
                    case SipTransports.UDP:
                        startUDPServer(targetTuple, keyHandler, instanceHandler);
                        startUDPClient(keyHandler, instanceHandler);
                        break;
                    case SipTransports.TCP:
                        startTCPServer(targetTuple, keyHandler, instanceHandler);
                        startTCPClient(keyHandler, instanceHandler);
                        break;
                    case SipTransports.TLS:
                        final ProtocolChain tlsProtocolChain = getTLSProtocolChain(ctx);
                        ProtocolChainInstanceHandler tlsInstanceHandler =
                                new SimpleProtocolChainInstanceHandler(tlsProtocolChain);
                        startTLSServer(targetTuple, keyHandler, tlsInstanceHandler);
                        startTLSClient(keyHandler, tlsInstanceHandler);
                        break;
                    case SipTransports.UNDEFINED://Hmm, should not happen
                        break;
                }
            }
        } else {
            if (logger.isLoggable(Level.FINE)) {
                logger.infoMsg("SipBindingCtx [" + ctx + "] is stale, not starting ctx");
            }
        }
    }

    private void restartCtxServers(String ctx, DefaultSelectionKeyHandler keyHandler) {
        //TODO
        if (logger.logWarning()) {
            logger.warning("restartCtxServers not implemented.");
        }
    }

    private void stopCtxServers(String ctx, DefaultSelectionKeyHandler keyHandler) {
        //TODO
        if (logger.logWarning()) {
            logger.warning("stopCtxServers not implemented.");
        }
    }

    private void startUDPServer(TargetTuple targetTuple,
            DefaultSelectionKeyHandler keyHandler,
            ProtocolChainInstanceHandler instanceHandler) {
        UDPSelectorHandler udpSelector = createUdpEndpoint(
                targetTuple.getSocketAddress());
        udpController.addSelectorHandler(udpSelector);
        udpSelector.setSelectionKeyHandler(keyHandler);
        udpController.setProtocolChainInstanceHandler(instanceHandler);
        udpController.setPipeline(serverPipeline);

    }

    private void startTCPServer(TargetTuple targetTuple,
            DefaultSelectionKeyHandler keyHandler,
            ProtocolChainInstanceHandler instanceHandler) {
        TCPSelectorHandler tcpSelector = createTcpEndpoint(
                targetTuple.getSocketAddress());
        tcpController.addSelectorHandler(tcpSelector);
        tcpSelector.setSelectionKeyHandler(keyHandler);
        tcpController.setProtocolChainInstanceHandler(instanceHandler);
        tcpController.setPipeline(serverPipeline);
    }

    private void startTLSServer(TargetTuple targetTuple,
            DefaultSelectionKeyHandler keyHandler,
            ProtocolChainInstanceHandler tlsInstanceHandler) {
        SSLSelectorHandler tlsSelector = createTLSEndpoint(
                targetTuple.getSocketAddress());
        tlsController.addSelectorHandler(tlsSelector);
        tlsSelector.setSelectionKeyHandler(keyHandler);
        tlsController.setProtocolChainInstanceHandler(tlsInstanceHandler);
        tlsController.setPipeline(serverPipeline);
    }

    private void startUDPClient(DefaultSelectionKeyHandler keyHandler,
            ProtocolChainInstanceHandler instanceHandler) {
        udpClientController.setProtocolChainInstanceHandler(instanceHandler);
        UDPSelectorHandler udpClientSelectorHandler = new UDPSelectorHandler(true);
        udpClientSelectorHandler.setSelectionKeyHandler(keyHandler);
        udpClientController.addSelectorHandler(udpClientSelectorHandler);
        udpClientController.setPipeline(clientPipeline);
    }

    private void startTCPClient(DefaultSelectionKeyHandler keyHandler,
            ProtocolChainInstanceHandler instanceHandler) {
        tcpClientController.setProtocolChainInstanceHandler(instanceHandler);
        TCPSelectorHandler tcpClientSelectorHandler = new TCPSelectorHandler(true);
        tcpClientSelectorHandler.setSelectionKeyHandler(keyHandler);
        tcpClientController.addSelectorHandler(tcpClientSelectorHandler);
        tcpClientController.setPipeline(clientPipeline);
    }

    private void startTLSClient(DefaultSelectionKeyHandler keyHandler,
            ProtocolChainInstanceHandler tlsInstanceHandler) {
        tlsClientController.setProtocolChainInstanceHandler(tlsInstanceHandler);
        TCPSelectorHandler tlsClientSelectorHandler = new SSLSelectorHandler(true);
        tlsClientSelectorHandler.setSelectionKeyHandler(keyHandler);
        tlsClientController.addSelectorHandler(tlsClientSelectorHandler);
        tlsClientController.setPipeline(clientPipeline);
    }

    public synchronized void stop() throws IOException {
        tcpController.stop();
        udpController.stop();
        tlsController.stop();
        tcpClientController.stop();
        udpClientController.stop();
        tlsClientController.stop();
    }

    private ProtocolChain getProtocolChain() {
        DefaultProtocolChain protocolChain = new DefaultProtocolChain();
        protocolChain.setContinuousExecution(false);

        ReadFilter readFilter = new SharedReadFilter();
        readFilter.setContinuousExecution(false);

        protocolChain.addFilter(readFilter);
        protocolChain.addFilter(new MessageProcessorFilter(this));

        return protocolChain;
    }

    private ProtocolChain getTLSProtocolChain(String ctx) {
        SSLReadFilter sslReadFilter = new ClientCertSSLReadFilter();
        sslReadFilter.setSSLContext(getSSLContext(ctx));
        Boolean clientAuth = Boolean.parseBoolean(
                ctxs.get(ctx).getSSLAttribute("ClientAuthEnabled"));
        sslReadFilter.setNeedClientAuth(clientAuth);

        ProtocolChain tlsProtocolChain = new DefaultProtocolChain();
        tlsProtocolChain.addFilter(sslReadFilter);
        tlsProtocolChain.addFilter(new MessageProcessorFilter(this));

        return tlsProtocolChain;
    }

    private SSLContext getSSLContext(String ctx) {
        ServerSocketFactory serverSF = null;
        try {

            // Workaround issue 213
            if (keyStore == null) {
                logger.warning("Invalid keyStore:" + keyStore + " Using default.");
                keyStore = System.getProperty("javax.net.ssl.keyStore");
            }
            // Workaround issue 213
            if (trustStore == null) {
                logger.warning("Invalid trustStore:" + trustStore + " Using default.");
                trustStore = System.getProperty("javax.net.ssl.trustStore");
            }

            //TODO This expects that the ServerSocketFactory initialized
            //is a seperated instance. A brief check of the grizzly
            //classes involved reveals that it is for the default JSSE14 factories
            SSLImplementation sslHelper = SSLImplementation.getInstance();
            serverSF = sslHelper.getServerSocketFactory();
            serverSF.setAttribute("keystoreType", "JKS");
            serverSF.setAttribute("keystore", keyStore);
            serverSF.setAttribute("truststoreType", "JKS");
            serverSF.setAttribute("truststore", trustStore);
            String keyAlias = ctxs.get(ctx).getSSLAttribute("CertNickname");
            serverSF.setAttribute("keyAlias", (keyAlias != null ? keyAlias : "s1as")); //Default GF s1sa
            serverSF.init();
        } catch (IOException e) {
        //TODO Logging
        } catch (ClassNotFoundException e) {
        //TODO Logging
        }
        return serverSF.getSSLContext();
    }

    void initPipeline() {
        serverPipeline = createPipeline(SERVER_PIPELINE_NAME, DEFAULT_PIPELINE_DISTINCTION_PORT);
        clientPipeline = clientsShareServerThreadPool ? serverPipeline : createPipeline(CLIENT_PIPELINE_NAME, DEFAULT_PIPELINE_DISTINCTION_PORT);

        SipContainerThreadPool tp = SipContainerThreadPool.getInstance();
        tp.initialize(serverPipeline);
    }

    private void reinitPipeline() {
        //TODO reinit the pipeline according to the reconfigured settings.
        if (logger.logWarning()) {
            logger.warning("reinitPipeline not implemented.");
        }
    }

    private DefaultPipeline createPipeline(String pipelineName, int distinctionPort) {
        DefaultPipeline pipeline = new DefaultPipeline(maxPipelineThreads,
                minPipelineThreads, pipelineName, distinctionPort);
        pipeline.setThreadsIncrement(pipelineThreadsIncrement);
        pipeline.setInitialByteBufferSize(pipelineInitialByteBufferSize);
        pipeline.setByteBufferType(ByteBufferType.HEAP);
        // No limits for now.
        pipeline.setQueueSizeInBytes(-1);

        return pipeline;
    }

    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);

            if (SipMonitoring.isEnabled(SipMonitoring.NETWORK_MANAGER)) {
                incrEasSentSipRequests();
            }


        } catch (IOException e) {
            if (logger.logSevere()) {
                logger.severe(e, "sip.stack.network.failed_to_send_request", req.toDebugString());
            }
        }
    }

    public void dispatch(SipServletResponseImpl resp) { //Failback, StreamResponseDispatchern is the normal path
        try {
            connectionManager.send(resp);

            if (SipMonitoring.isEnabled(SipMonitoring.NETWORK_MANAGER)) {
                incrEasSentSipResponses();
            }

        } catch (IOException e) {
            try{
                TargetTuple newTupple = TargetResolver.getInstance().resolveResponse(resp);
                if (newTupple != null && newTupple != resp.getRemote()) {
                    resp.setRemote(newTupple);

                    // Cannot use recurrance here because it will loop.
                    connectionManager.send(resp);

                    if (SipMonitoring.isEnabled(SipMonitoring.NETWORK_MANAGER)) {
                        incrEasSentSipResponses();
                    }                
                }
            } catch (Exception e2) {   
                if (logger.logSevere()) {
                    logger.severe(e, "sip.stack.network.failed_to_send_response", resp.toDebugString());
                }
                // Only report the last exception
                return;
            }
                
            if (logger.logSevere()) {
                logger.severe(e, "sip.stack.network.failed_to_send_response", resp.toDebugString());
            }
        }
    }

    TCPSelectorHandler createTcpEndpoint(InetSocketAddress addr) {
        int linger = -1;

        final TCPSelectorHandler selectorHandler = new TCPSelectorHandler(){            
            /**
             * Intercept the accept operations and cache the associated connection.
             */
            @Override
            public boolean onAcceptInterest(SelectionKey key,
                    Context ctx) throws IOException{     
                
                SelectableChannel channel = acceptWithoutRegistration(key);        
                if (channel != null) {
                    configureChannel(channel);
                    SelectionKey readKey =
                            channel.register(selector, SelectionKey.OP_READ);
                    // Cache the connection.
                    connectionManager.add(readKey, this); 
                }
                return false;
            }              
        };
        selectorHandler.setPort(addr.getPort());
        selectorHandler.setInet(addr.getAddress());
        selectorHandler.setLinger(linger);
        selectorHandler.setLogger(logger.getLogger());
        selectorHandler.setReuseAddress(true);
        selectorHandler.setSocketTimeout(keepAliveTimeoutInSeconds * 1000);
        return selectorHandler;
    }

    SSLSelectorHandler createTLSEndpoint(InetSocketAddress addr) {
        int linger = -1;

        SSLSelectorHandler selectorHandler = new SSLSelectorHandler();
        selectorHandler.setPort(addr.getPort());
        selectorHandler.setInet(addr.getAddress());
        selectorHandler.setLinger(linger);
        selectorHandler.setLogger(logger.getLogger());
        selectorHandler.setReuseAddress(true);
        selectorHandler.setSocketTimeout(keepAliveTimeoutInSeconds * 1000);
        return selectorHandler;
    }

    UDPSelectorHandler createUdpEndpoint(InetSocketAddress addr) {
        int linger = -1;

        UDPSelectorHandler selectorHandler = new UDPSelectorHandler();
        selectorHandler.setPort(addr.getPort());
        selectorHandler.setInet(addr.getAddress());
        selectorHandler.setLogger(logger.getLogger());
        selectorHandler.setReuseAddress(true);
        return selectorHandler;
    }

    public void run() {
        try {
            // TODO: Use a thread pool
            new Thread() {

                @Override
                public void run() {
                    try {
                        tcpClientController.start();
                    } catch (IOException ex) {
                        if (logger.logSevere()) {
                            logger.severe(ex, "sip.stack.network.controller_death", "tcp", "client", ex);
                        }
                    }
                }
            }.start();
            new Thread() {

                @Override
                public void run() {
                    try {
                        udpClientController.start();
                    } catch (IOException ex) {
                        if (logger.logSevere()) {
                            logger.severe(ex, "sip.stack.network.controller_death", "udp", "client", ex);
                        }
                    }
                }
            }.start();
            new Thread() {

                @Override
                public void run() {
                    try {
                        udpController.start();
                    } catch (IOException ex) {
                        if (logger.logSevere()) {
                            logger.severe(ex, "sip.stack.network.controller_death", "udp", "server", ex);
                        }
                    }
                }
            }.start();
            new Thread() {

                @Override
                public void run() {
                    try {
                        tlsController.start();
                    } catch (IOException ex) {
                        if (logger.logSevere()) {
                            logger.severe(ex, "sip.stack.network.controller_death", "tls", "server", ex);
                        }
                    }
                }
            }.start();
            new Thread() {

                @Override
                public void run() {
                    try {
                        tlsClientController.start();
                    } catch (IOException ex) {
                        if (logger.logSevere()) {
                            logger.severe(ex, "sip.stack.network.controller_death", "tls", "client", ex);
                        }
                    }
                }
            }.start();

            tcpController.start();
        } catch (IOException e) {
            if (logger.logSevere()) {
                logger.severe(e, "sip.stack.network.controller_death", "tcp", "server", e);
            }
        }
    }

    class ConnectionManager {

        private int maxCachedBB = 50;
        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 = maxCachedByteBuffers;
        }
        
        private void add(SelectionKey key, TCPSelectorHandler selectorHandler){                        
            SocketChannel socketChannel = (SocketChannel)key.channel();
            
            // TODO Cache object instead of invoking new on every request.
            TCPOutboundConnectorHandler outboundConnector = new TCPOutboundConnectorHandler();
            outboundConnector.setSelectorHandler(selectorHandler);
            outboundConnector.setUnderlyingChannel(socketChannel);

            Socket s = socketChannel.socket();
            InetSocketAddress remoteAddress = (InetSocketAddress) s.getRemoteSocketAddress();
            TargetTuple remote = new TargetTuple(SipTransports.TCP_PROT, remoteAddress);
            InetSocketAddress local = (InetSocketAddress) s.getLocalSocketAddress(); 
            
            outboundConnector.setCallbackHandler(createTCPCallbackHandler(outboundConnector, remote));
            
            // Bind the Tupple to the SelectionKey so when the connection is closed,
            // The TCPOutboundConnectorHandler is removed from the cache.
            // TODO: Add support SSL.
            key.attach(remote);
            streams.put(remote, outboundConnector);
        }
        
        private void remove(SelectionKey key){
            Object o = key.attachment();
            if (o != null && o instanceof TargetTuple){
                streams.remove((TargetTuple)o);
            }
        }
                
        private void send(SipServletMessageImpl sipServletMessage)
                throws IOException {
            send(sipServletMessage, true, null);
        }

        private void send(SipServletMessageImpl sipServletMessage,
                boolean useCache, ByteBuffer bb) throws IOException {
            
            if (logger.isLoggable(Level.FINE)) {
                if( sipServletMessage.getMessageType() == SipServletMessageImpl.SipMessageType.SipRequest ) {
                    logger.logMsg(Level.FINE, "Network OUT request --> \r\n" + sipServletMessage.toString());
                } else {
                    logger.logMsg(Level.FINE, "Network OUT response --> \r\n" + sipServletMessage.toString());
                }
            }

            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) {
                if (handler instanceof TCPConnectorHandler){
                    isConnected = ((TCPConnectorHandler) handler).isConnected();
                } else {
                    isConnected = true;
                }
            } 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) {
                //[Eltjo] This code looks fishy, note that once socketBufferSize
                // is set to socket buf size, it remains for all send calls after!
                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) {
                    }
                }
                sipServletMessage.toBufferInit();
                bb = toBuffer(sipServletMessage, socketBufferSize);
            }

            if (bb.position() == bb.limit() && bb.position() == 0) {
                if (logger.logSevere()) {
                    logger.severe("sip.stack.network.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) {
                                if (logger.logWarning()) {
                                    logger.warning(ex, "sip.stack.network.connection_write_failed");
                                }
                            }
                        }
                        long nWrite = handler.write(bb, blocking);
                        while (sipServletMessage.toBufferHasRemaining()) {
                            //This is a big message
                            bb = toBuffer(sipServletMessage, socketBufferSize);
                            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);
                            while (sipServletMessage.toBufferHasRemaining()) {
                                //This is a big message
                                bb = toBuffer(sipServletMessage, socketBufferSize);
                                bbs.offer(bb);
                            }
                            if (wasNull) {
                                asyncWriteQueue.put(tt, bbs);
                            }
                        } else {
                            if (logger.logWarning()) {
                                logger.warning("sip.stack.network.unable_to_connect", "\n" + tt);
                            }
                        }
                    }
                } catch (IOException ex) {
                    if (logger.isLoggable(Level.FINE)) {
                        logger.logMsg(Level.FINE, ex, "Unable to write");
                    }

                    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;
                        //Changed to null since message can be multiple buffers
                        send(sipServletMessage, false, null); 
                        return;
                    } else {
                        throw ex;
                    }
                } finally {
                    writeLock.unlock();
                }
            } 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 final CallbackHandler createTCPCallbackHandler
                    (final ConnectorHandler connectorHandler, final TargetTuple tt){
            CallbackHandler<Context> callbackHandler = new CallbackHandler<Context>() {

                public void onConnect(IOEvent<Context> ioEvent) {
                    SelectionKey key = ioEvent.attachment().getSelectionKey();
                    try {
                        connectorHandler.finishConnect(key);
                        tcpClientController.registerKey(key, SelectionKey.OP_READ,
                                Controller.Protocol.TCP);
                    } catch (IOException ex) {
                        if (logger.isLoggable(Level.FINE)) {
                            logger.logMsg(Level.FINE, ex, "Connect failed" 
                                        + tt.getSocketAddress());
                        }
                        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) {
                        if (logger.logSevere()) {
                            logger.severe(e, "sip.stack.network.connection_read_failed", "tcp");
                        }
                    }
                }

                public void onWrite(IOEvent<Context> ioEvent) {
                    throw new IllegalStateException("Should not end up here!");
                }
            };
            return callbackHandler;
        }
        
        
        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;
            }

            try {
                connectorHandler.connect(remote, createTCPCallbackHandler(connectorHandler,tt));
                return connectorHandler;
            } catch (Throwable t) {
                if (logger.isLoggable(Level.FINE)) {
                    logger.logMsg(Level.FINE, t, "Connect failed:" + remote);
                }
                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 {
                        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.logMsg(Level.FINE, ex, "Connect failed:" + remote);
                        }
                        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) {
                        if (logger.logSevere()) {
                            logger.severe(e, "sip.stack.network.connection_read_failed", "tls");
                        }
                    }
                }

                // 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.logMsg(Level.FINE, t, "Connect failed:" + remote);
                }
                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.logMsg(Level.FINE, ex, "Connect failed:" + remote);
                        }
                        streams.remove(tt);
                        udpClientController.getSelectorHandler(Protocol.UDP).getSelectionKeyHandler().cancel(key);
                    }
                }

                public void onRead(IOEvent<Context> ioEvent) {
                    try {
                        Context ctx = ioEvent.attachment();
                        // Add a flag to force Grizzly to close the connection 
                        // when there is no more bytes availables.
                        ctx.setAttribute(IS_CLIENT_EXECUTION,IS_CLIENT_EXECUTION);
                        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) {
                        if (logger.logSevere()) {
                            logger.severe(e, "sip.stack.network.connection_read_failed", "udp");
                        }
                    }
                }

                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) {
                        if (logger.logSevere()) {
                            logger.severe(ex, "sip.stack.network.connection_write_failed_udp");
                        }
                    } 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.logMsg(Level.FINE, t, "Connect failed:" + remote);
                }
                return null;
            }
        }
    }

    class SharedReadFilter extends ReadFilter {

        private int bufferSize = 16 * 8192;

        @Override
        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) {
                    if (logger.logWarning()) {
                        logger.warning(ex, "sip.stack.network.resize_buffer_failed_bad_Socket_state");
                    }
                    bufferSize = 16 * 8192;
                }
                workerThread.setByteBuffer(ByteBuffer.allocate(bufferSize));
            }
            return super.execute(ctx);
        }
        
        @Override
        public boolean postExecute(Context ctx) throws IOException {
            String isClientExecution = 
                    (String)ctx.removeAttribute(IS_CLIENT_EXECUTION);
            // isClientExecution will be null when this filter is used
            // as a server side component, and not null when used from a 
            // client side component. When used by a client side component, 
            // do not override the decision of closing or not the SelectionKey
            if (isClientExecution == null 
                    && ctx.getProtocol() == Controller.Protocol.UDP 
                    && ctx.getKeyRegistrationState() != Context.KeyRegistrationState.REGISTER) {
                // Main UDP listener should never be closed!
                ctx.setKeyRegistrationState(Context.KeyRegistrationState.REGISTER);
            }

            return super.postExecute(ctx);
        }
    }


    class ClientCertSSLReadFilter extends SSLReadFilter {
        @Override
        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;
        }
    }
    
    class SimpleProtocolChainInstanceHandler implements ProtocolChainInstanceHandler {
        ProtocolChain protocolChain = null;

        public SimpleProtocolChainInstanceHandler(ProtocolChain aProtocolChain) {
            protocolChain = aProtocolChain;
        }

        public ProtocolChain poll() {
            return protocolChain;
        }

        public boolean offer(ProtocolChain instance) {
            return true;
        }
    }

    //secure
    @Configuration(key = "javax.net.ssl.keyStore", update = UpdatePolicy.STARTUP)
    public void setSslKeyStore(String aKeyStore) {
        keyStore = aKeyStore;
    }

    @Configuration(key = "javax.net.ssl.trustStore", update = UpdatePolicy.STARTUP)
    public void setSslTrustStore(String aTrustStore) {
        trustStore = aTrustStore;
    }

    //grizzly pipelines and pools
    @Configuration(key = "TimeoutInSeconds", node = "/SipService/KeepAlive", update = UpdatePolicy.STARTUP, usage = UsagePolicy.IGNORE)
    public void setKeepAliveTimeoutInSeconds(int aKeepAliveTimeoutInSeconds) {
        keepAliveTimeoutInSeconds = aKeepAliveTimeoutInSeconds;
    }

    @Configuration(key = "ThreadCount", node = "/SipService/RequestProcessing", usage = UsagePolicy.IGNORE)
    public void setMaxPipelineThreads(int aMaxThreads) {
        maxPipelineThreads = aMaxThreads;
        reinitPipeline();
    }

    @Configuration(key = "sip.network.grizzly.useServerThreadPool", update = UpdatePolicy.STARTUP)
    public void setClientsShareServerThreadPool(boolean isShared) {
        clientsShareServerThreadPool = isShared;
    }

    //Is checked with Jeanfrancios, can be mapped to the initial-thread-count.
    //This means that the min and initial are the same!
    @Configuration(key = "InitialThreadCount", node = "/SipService/RequestProcessing", usage = UsagePolicy.IGNORE)
    public void setMinPipelineThreads(int aMinThreads) {
        minPipelineThreads = aMinThreads;
        reinitPipeline();
    }

    @Configuration(key = "ThreadIncrement", node = "/SipService/RequestProcessing", usage = UsagePolicy.IGNORE)
    public void setPipelineThreadsIncrement(int aThreadsIncrement) {
        pipelineThreadsIncrement = aThreadsIncrement;
    }

    @Configuration(key = "HeaderBufferLengthInBytes", node = "/SipService/RequestProcessing", usage = UsagePolicy.IGNORE)
    public void setPipelineInitialByteBufferSize(int anInitialByteBufferSize) {
        pipelineInitialByteBufferSize = anInitialByteBufferSize;
    }

    //grizzly buffers
    //@Configuration (key="sip.network.grizzly.maxCachedByteBuffer", update=UpdatePolicy.STARTUP)    
    @Configuration(key = "MaxPendingCount", node = "/SipService/ConnectionPool", usage = UsagePolicy.IGNORE)
    public void setMaxCachedByteBuffers(int aMaxCachedByteBuffers) {
        maxCachedByteBuffers = aMaxCachedByteBuffers;
    }
}
