/*
 * 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.Constants;
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
 */
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 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 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 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.
     */
    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
        _selector = Selector.open();
        _UDPListener = new UDPListener(Integer.parseInt(_config.get(
                        Constants.SIP_PORT)), this, _nextLayer);
        _UDPListener.open();
        _TCPListener = new TCPListener(Integer.parseInt(_config.get(
                        Constants.SIP_PORT)), this, _nextLayer, true);
        _TCPListener.open();
    }

    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
     */
    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.run();
                            } 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();
                                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);
            }
        }
    }

    /**
     * Executes the Runnable in a thread pool of worker threads
     */

    // TODO consider removing this method and use
    // SipContainerThreadPool.execute() directly.....
    public void execute(final Runnable r) {
        _threadPool.execute(r);
    }

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

    // package access
    void incrEasInvalidSipMessages() {
        _EasInvalidSipMessages.incrementAndGet();
    }

    void incrEasReceivedSipRequests() {
        _EasReceivedSipRequests.incrementAndGet();
    }

    void incrEasReceivedSipResponses() {
        _EasReceivedSipResponses.incrementAndGet();
    }

    void incrEasSentSipRequests() {
        _EasSentSipRequests.incrementAndGet();
    }

    void incrEasSentSipResponses() {
        _EasSentSipResponses.incrementAndGet();
    }

    public long getEasInvalidSipMessages() {
        return _EasInvalidSipMessages.longValue();
    }

    public long getEasReceivedSipRequests() {
        return _EasReceivedSipRequests.longValue();
    }

    public long getEasReceivedSipResponses() {
        return _EasReceivedSipResponses.longValue();
    }

    public long getEasSentSipRequests() {
        return _EasSentSipRequests.longValue();
    }

    public long getEasSentSipResponses() {
        return _EasSentSipResponses.longValue();
    }

    /**
     * Defines timeout value in milliseconds for a single write operation of a
     * sip link.
     * </p>
     * Range: 1-50ms
     *
     * @param timeout
     */
    public void setSipLinkTimeout(int timeout) {
        if ((timeout > 0) && (timeout <= 50)) {
            _SipLinkTimeout = timeout;
        } else {
            _log.log(Level.WARNING,
                "Bad SipLinkTimeout = " + timeout +
                ", 0 < select <= 50 allowed");
        }
    }

    /**
     * Defines timeout value in milliseconds for a single write operation of a
     * sip link.
     *
     * @return timeout for write operation of a sip link
     */
    public int getSipLinkTimeout() {
        return _SipLinkTimeout;
    }

    /**
     * Defines timeout value in seconds for keeping an inactive link. If no
     * traffic has been generated on that link for that time period the link is
     * closed.
     *
     * @return timeout value in seconds for keeping an inactive link
     */
    public long getSipLinkAliveTimeout() {
        return _SipLinkAliveTimeout;
    }

    /**
     * Defines timeout value in seconds for keeping an inactive link. If no
     * traffic has been generated on that link for that time period the link is
     * closed.
     *
     * @param timeout
     *           value in seconds for keeping an inactive link
     */
    public void setSipLinkAliveTimeout(long timeout) {
        _SipLinkAliveTimeout = timeout;
    }

    /**
     * Defines the number of retries to perform of a single write operation of a
     * sip link.
     * </p>
     * Range: 1-25
     * </p>
     * The maximum time a link will try for success of a write operation is
     * defined by SipLinkTimeout * SipLinkTimeoutRetries milliseconds.
     *
     * @param retries
     */
    public void setSipLinkTimeoutRetries(int retries) {
        if ((retries > 0) && (retries <= 25)) {
            _SipLinkTimeoutRetries = retries;
        } else {
            _log.log(Level.WARNING,
                "Bad SipLinkTimeoutRetries = " + retries +
                ", 0 < select <= 25 allowed");
        }

        _SipLinkTimeoutRetries = retries;
    }

    /**
     * Defines the number of retries to perform of a single write operation of a
     * sip link.
     * </p>
     * The maximum time a link will try for success of a write operation is
     * defined by SipLinkTimeout * SipLinkTimeoutRetries milliseconds.
     *
     * @return the number of retries to perform of a single write operation of a
     *         sip link.
     */
    public int getSipLinkTimeoutRetries() {
        return _SipLinkTimeoutRetries;
    }

    /**
     * Returns the maximum number of connect/write tasks that can be queued in a
     * link.
     *
     * @return the maximum number of connect/write tasks that can be queued in a
     *         link
     */
    public int getSipLinkMaxQueueLength() {
        return _SipLinkMaxQueueLen;
    }

    /**
     * Returns the maximum number of connect/write tasks that can be queued in a
     * link.
     *
     * @param length
     *           the maximum number of connect/write tasks that can be queued in
     *           a link
     */
    public void setSipLinkMaxQueueLength(int length) {
        _SipLinkMaxQueueLen = length;
    }

    /**
     * Returns the maximum time a thread will wait to get an exclusive lock for a
     * sip link.
     *
     * @return the maximum time a thread will wait to get an exclusive lock for a
     *         sip link.
     */
    public int getSipLinkWaitLockTimeout() {
        return _SipLinkWaitLockTimeout;
    }

    /**
     * Sets the maximum time a thread will wait to get an exclusive lock for a
     * sip link.
     *
     * @param timeout
     */
    public void setSipLinkWaitLockTimeout(int timeout) {
        _SipLinkWaitLockTimeout = timeout;
    }

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

    /**
     * Returns whether a malformed request should be handled by sending a proper
     * informational error response.
     *
     * @return
     */
    public Boolean isErrorResponseEnabled() {
        return _errorResponseEnabled;
    }

    /**
     * Setter of the <code>ErrorResponseEnabled</code> property.
     *
     * @param responseEnabled
     */
    public void setErrorResponseEnabled(Boolean responseEnabled) {
        _errorResponseEnabled = responseEnabled;
    }
}
