/*
 * 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.ericsson.ssa.config.Config;
import com.ericsson.ssa.config.ConfigFactory;
import com.ericsson.ssa.config.annotations.Configuration;
import com.ericsson.ssa.sip.Dispatcher;
import com.ericsson.ssa.sip.Layer;
import com.ericsson.ssa.sip.LayerHelper;
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 java.io.IOException;

import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;

// inserted by hockey (automatic)
import java.util.logging.Logger;


/**
 * @author ekrigro
 * @reveiwed ehsroha 2006-nov-14
 * @reviewed qmigkra 2006-nov-20
 * @reviewed edtreil 2007-feb-19
 */
//Because the OldNetworkManager is not context aware the first
//suitable context is taken! That is the first with a udp_tcp transport configured.
//This is not ideal as the order of the context is undetermined.
//Therefore configure only 1 suitable context when using the OldNetworkManager!
public class OLDNetworkManager extends NetworkManager implements Runnable {
    /**
     * singleton instance
     */
    private static OLDNetworkManager _singletonInstance = new OLDNetworkManager();
    protected Thread _activeThread = null;
    protected boolean _stopped = true;
    protected Selector _selector = null;
    private UDPListener _UDPListener = null;
    private TCPListener _TCPListener = null;
    private UDPListener _UDPSubsequentListener = null;
    private TCPListener _TCPSubsequentListener = null;
    private Logger _log = (Logger) Logger.getLogger("SipContainer");

    //private LocalStringManager localStrings = new LocalStringManager( NetworkManager.class.getName() );
    private Layer _nextLayer = null;
    private SipContainerThreadPool _threadPool = SipContainerThreadPool.getInstance();

    /**
     * Reference to the dispatcher of the SIP message, to process the message
     * after parsing
     */
    private Dispatcher dispatcher = null;
    private int _threadPoolSize = 10;
    private Map<TargetTuple, Link> _activeConnections = new ConcurrentHashMap<TargetTuple, Link>(20);
    private Queue<Link> _registerConnections = new LinkedBlockingQueue<Link>();
    private List<Link> _acceptRegConnections = new ArrayList<Link>();
    private Object _acceptConnectionsLock = new Object();
    private Semaphore _sem = new Semaphore(0);
    private int _SipLinkTimeout = 10; // ms
    private int _SipLinkTimeoutRetries = 25; // iterations
    private int _SipLinkMaxQueueLen = 50;
    private int _SipLinkWaitLockTimeout = 5000;
    private long _SipLinkAliveTimeout = 120;
    private boolean _errorResponseEnabled = false;
    private AtomicLong _EasInvalidSipMessages = new AtomicLong();
    private AtomicLong _EasReceivedSipRequests = new AtomicLong();
    private AtomicLong _EasReceivedSipResponses = new AtomicLong();
    private AtomicLong _EasSentSipRequests = new AtomicLong();
    private AtomicLong _EasSentSipResponses = new AtomicLong();
    private Config _config = ConfigFactory.getConfig();

    void acquireSelectorSemaphore() {
        _sem.acquireUninterruptibly();
    }

    void releaseSelectorSemaphore() {
        try {
            _sem.release();
        } catch (Throwable fatal) {
            _log.log(Level.SEVERE,
                "Unable to release semaphore of main read loop, restart server!",
                fatal);
        }
    }

    /**
     * Selector thread will register the link with read events.
     *
     * @param l
     *           link to register
     */
    public void registerConnection(Link l) {
        _registerConnections.add(l);
    }

    /**
     * Returns an existing link or creates a new link and returns it if it
     * doesn't exist.
     *
     * @param tt
     * @return
     */
    public Link getActiveConnection(TargetTuple tt) throws IOException {
        if ((tt == null) || (tt.getProtocol() == SipTransports.UDP_PROT)) {
            //
            // udp
            //
            return _UDPListener;
        }

        //
        // tcp
        //
        Link link = _activeConnections.get(tt);

        if (link == null) {
            // to avoid multiple links to same destination
            synchronized (_activeConnections) {
                // double locked pattern
                link = _activeConnections.get(tt);

                if (link == null) {
                    // lets create a new tcp link
                    link = new TCPLink(tt, this, _nextLayer);
                    // store it for re-use by others
                    addActiveConnection(link);

                    if (_log.isLoggable(Level.FINE)) {
                        _log.log(Level.FINE,
                            "No TCP connection, opening new to : " +
                            link.getInfo());
                    }
                }
            }

            if (!link.isOpen()) {
                // connect to remote, might take a while...
                link.open();
            }
        } else {
            if (!link.isOpen()) {
                // connect to remote, might take a while...
                link.open();
            }

            if (_log.isLoggable(Level.FINE)) {
                _log.log(Level.FINE,
                    "Reusing TCP connection to : " + link.getInfo());
            }
        }

        return link;
    }

    public void addActiveConnection(Link l) {
        Link replaced = _activeConnections.put(l.getTargetTuple(), l);

        if (replaced != null) {
            if (_log.isLoggable(Level.INFO)) {
                _log.log(Level.INFO,
                    "Connected link already exist to destination: " +
                    l.getInfo() + ".\n Connected links = " +
                    _activeConnections.size());
            }
        }

        if (_log.isLoggable(Level.FINE)) {
            _log.log(Level.FINE,
                "Added connected link: " + l.getInfo() +
                ".\n Connected links = " + _activeConnections.size());
        }
    }

    public Link removeActiveConnection(Link l) {
        Link toReturn = _activeConnections.remove(l.getTargetTuple());

        if (toReturn != null) {
            if (_log.isLoggable(Level.FINE)) {
                _log.log(Level.FINE,
                    "Removed connected link: " + l.getInfo() +
                    ".\n Connected links = " + _activeConnections.size());
            }
        } else {
            if (_log.isLoggable(Level.INFO)) {
                _log.log(Level.INFO,
                    "Connected link already removed: " + l.getInfo() +
                    ".\n Connected links = " + _activeConnections.size());
            }
        }

        return toReturn;
    }

    public void addAcceptedConnection(Link l) {
        synchronized (_acceptConnectionsLock) {
            _acceptRegConnections.add(l);

            if (_log.isLoggable(Level.FINE)) {
                _log.log(Level.FINE,
                    "Added accepted link: " + l + ".\n Accepted links = " +
                    _acceptRegConnections.size());
            }
        }
    }

    public void removeAcceptedConnection(Link l) {
        synchronized (_acceptConnectionsLock) {
            boolean isFound = _acceptRegConnections.remove(l);

            if (isFound && _log.isLoggable(Level.FINE)) {
                _log.log(Level.FINE,
                    "Removed accepted link: " + l + ".\n Accepted link = " +
                    _acceptRegConnections.size());
            }
        }
    }

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

    /**
     * @param threadPoolSize
     *           The threadPoolSize to set.
     */
    @Configuration (key="thread-count", node="/SipService/RequestProcessing")
    public void setThreadPoolSize(Integer threadPoolSize) {
        if (threadPoolSize > 0) {
            _threadPoolSize = threadPoolSize;
        }
    }

    Selector getSelector() { // Helper method for links

        return _selector;
    }

    /**
     * Register the dispatcher of the message. This reference is passed up
     * through the stack
     */
    public void registerDispatcher(Dispatcher dispatcher) {
        this.dispatcher = dispatcher;
    }

    public Dispatcher getDispatcher() {
        return dispatcher;
    }

    public synchronized void initialize() throws IOException {
        if (isRunning()) {
            throw new IOException("Network already running.");
        }

        _threadPool.initialize(_threadPoolSize);
        debugParameters();
    }

    // ================== Bean Properties =====================
    public synchronized void start() throws Exception {
        if (isRunning()) {
            throw new IOException("Network already running.");
        }

        _stopped = false;

        if ((_threadPool != null) && _threadPool.isShutdown()) {
            // the stack has been stoped/restarted
            _threadPool.initialize(_threadPoolSize);
        }

        // Initialize network service
        SipBindingCtx ctx = fetchSuitableSipBindingCtx();
            
        if (ctx!=null) {
            _selector = Selector.open();

            for (TargetTuple targetTuple : ctx.getTargetTuples()) {
                switch(targetTuple.getProtocol().ordinal()) {
                case SipTransports.UDP:
                    _UDPListener = new UDPListener(targetTuple, this, _nextLayer);
                    _UDPListener.open();
                    break;
                case SipTransports.TCP:
                    _TCPListener = new TCPListener(targetTuple, this, _nextLayer, true);
                    _TCPListener.open();
                    break;
                }
            }
        } else {
            //TODO use the right method of reporting FAILURE in startup!
            //Localize message and exception.
            _log.severe("No SipBinding found containing UPD and TCP transport!");
            throw new IllegalStateException("Can't start OldNetworkManager: no suitable SipBinding context found! Check your sip-listener configuration.");
        }
    }

    public synchronized void stop() throws IOException {
        if (_stopped) {
            throw new IOException("Network is not running.");
        }

        _stopped = true;
        _selector.wakeup();

        while (_activeThread != null) {
            try {
                wait(1000);
            } catch (InterruptedException ie) { /* Ignore */
            }
        }

        Iterator<Link> i = _activeConnections.values().iterator();

        while (i.hasNext()) {
            Link l = i.next();
            l.close();
        }

        _activeConnections.clear();
        i = _registerConnections.iterator();

        while (i.hasNext()) {
            Link l = i.next();
            l.close();
        }

        _registerConnections.clear();
        // close all listeners
        _TCPListener.close();
        _UDPListener.close();
        _selector.close();
        _threadPool.shutdown();
    }

    /**
     * Checks if the network is ready for sending/receiving SipMessages
     *
     * @return
     */
    @Override
    public boolean isRunning() {
        return !_stopped;
    }

    public void run() {
        Link newLink = null;

        while (!_stopped) {
            try {
                if (_selector.select() > 0) {
                    for (Iterator<SelectionKey> i = _selector.selectedKeys()
                                                             .iterator();
                            i.hasNext();) {
                        SelectionKey key = i.next();
                        i.remove();

                        Link link = null;

                        if (!key.isValid()) {
                            if (_log.isLoggable(Level.FINE)) {
                                _log.log(Level.FINE, "Key is not valid " + key);
                            }

                            if ((link != null) && link.isOpen()) {
                                link.close();

                                if (_log.isLoggable(Level.FINE)) {
                                    _log.log(Level.FINE,
                                        "Closing link = " + link.getInfo());
                                }
                            }
                        } else if (key.isAcceptable()) {
                            if (_log.isLoggable(Level.FINE)) {
                                _log.log(Level.FINE, "Key isAcceptable()" +
                                    key);
                            }

                            try {
                                link = (Link) key.attachment();
                                // run in selector thread
                                link.call();
                            } catch (Exception e) {
                                key.cancel();

                                if (link != null) {
                                    link.close();
                                    _log.log(Level.SEVERE,
                                        "Closing link = " + link.getInfo(), e);
                                }
                            }
                        } else if (key.isReadable()) {
                            if (_log.isLoggable(Level.FINE)) {
                                _log.log(Level.FINE, "Key isReadable()" + key);
                            }

                            try {
                                link = (Link) key.attachment();
                                // run in pool thread
                                link.markAsActive();
                                SipContainerThreadPool.getInstance().execute(link);
                                // wait for semaphore to be released
                                acquireSelectorSemaphore();
                            } catch (Exception e) {
                                key.cancel();

                                if (link != null) {
                                    link.close();
                                    _log.log(Level.SEVERE,
                                        "Closing link = " + link.getInfo(), e);
                                }
                            }
                        } else if (key.isWritable()) {
                            if (_log.isLoggable(Level.FINE)) {
                                _log.log(Level.FINE, "Key isWritable()" + key);
                            }
                        }
                    }
                }

                //
                // have to register keys with selector thread
                //
                while ((newLink = _registerConnections.poll()) != null) {
                    try {
                        SelectableChannel sc = newLink.getSelectableChannel();

                        if (sc.isOpen()) {
                            if (_log.isLoggable(Level.INFO)) {
                                _log.log(Level.INFO,
                                    "Adding new open channel : " +
                                    newLink.getInfo() +
                                    ", current number of keys in selector = " +
                                    _selector.keys().size());
                            }

                            sc.register(_selector, SelectionKey.OP_READ, newLink);
                        }
                    } catch (Exception e) {
                        _log.log(Level.WARNING, "Failed to register link.", e);
                    }
                }
            } catch (IOException ioe) {
                _log.log(Level.SEVERE, "", ioe);
            } catch (RuntimeException re) {
                _log.log(Level.SEVERE, "", re);
            }
        }
    }

    public void next(SipServletRequestImpl req) {
        req.pushTransactionDispatcher(this);
        req.pushApplicationDispatcher(this);
        LayerHelper.next(req, this, _nextLayer);
    }

    public void next(SipServletResponseImpl resp) {
        LayerHelper.next(resp, this, _nextLayer);
    }

    /**
     * Register the dispatcher of the message. This reference is passed up
     * through the stack
     */
    public void registerNext(Layer layer) {
        _nextLayer = layer;
    }

    public void dispatch(SipServletRequestImpl req) {
        try {
            Link link = getActiveConnection(req.getRemote());

            if (link != null) {
                link.dispatch(req);
            } else {
                if (_log.isLoggable(Level.INFO)) {
                    _log.log(Level.INFO,
                        "Failed to get link to + " + req.getRemote() +
                        ", dropping request " + req.toString());
                }
            }
        } catch (IOException ioe) {
            if (_log.isLoggable(Level.FINE)) {
                _log.log(Level.FINE, "" + req.getRemote(), ioe);
            }
        }
    }

    public void dispatch(SipServletResponseImpl resp) {
        try {
            Link link = getActiveConnection(resp.getRemote());

            if (link != null) {
                link.dispatch(resp);
            } else {
                if (_log.isLoggable(Level.INFO)) {
                    _log.log(Level.INFO,
                        "Failed to get Link to + " + resp.getRemote() +
                        ", dropping response " + resp.toString());
                }
            }
        } catch (IOException ioe) {
            if (_log.isLoggable(Level.INFO)) {
                _log.log(Level.INFO, "" + resp.getRemote(), ioe);
            }
        }
    }

    public static OLDNetworkManager getInstance() {
        return _singletonInstance;
    }




    private void debugParameters() {
        if (_log.isLoggable(Level.FINE)) {
            _log.log(Level.FINE, "sipLinkTimeout = " + getSipLinkTimeout());
        }

        if (_log.isLoggable(Level.FINE)) {
            _log.log(Level.FINE,
                "sipLinkTimeoutRetries = " + getSipLinkTimeoutRetries());
        }

        if (_log.isLoggable(Level.FINE)) {
            _log.log(Level.FINE,
                "sipLinkWaitLockTimeout = " + getSipLinkWaitLockTimeout());
        }

        if (_log.isLoggable(Level.FINE)) {
            _log.log(Level.FINE,
                "sipLinkMaxThreadQueueLength = " + getSipLinkMaxQueueLength());
        }
    }
    
    private String currentLocalHost = null;
    public String getCurrentLocalHost() {
        if (currentLocalHost==null) {
            currentLocalHost = fetchSuitableSipBindingCtx().getTargetTuples()[0].getIP();
        }
        
        return currentLocalHost;
    }
    
    private SipBindingCtx fetchSuitableSipBindingCtx() {
        SipBindingCtx suitableCtx = null;
        for (String ctx : SipBindingResolver.instance().getLocalContexts()) {
            suitableCtx = SipBindingResolver.instance().getContext(ctx);
            boolean hasUDP = false;
            boolean hasTCP = false;

            for (TargetTuple targetTuple : suitableCtx.getTargetTuples()) {
                hasUDP = hasUDP || targetTuple.getProtocol().ordinal() == SipTransports.UDP;
                hasTCP = hasTCP || targetTuple.getProtocol().ordinal() == SipTransports.TCP;
            }

            if (hasUDP && hasTCP) {
                break; //Found one.
            } else {
                suitableCtx = null;
            }
        }

        return suitableCtx;
    }
}
