/*
 * 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.container.processor.QueuedTask;
import com.ericsson.ssa.sip.Layer;
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.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketException;

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

import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;

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


/**
 * @author ehsroha
 * @reveiwed ehsroha 2006-nov-14
 */
public class TCPLink extends LinkBase implements Link {
    private Logger _log = (Logger) Logger.getLogger("SipContainer");
    private final TargetTuple _targetTuple;
    private SocketChannel _socketChannel = null;
    private InetSocketAddress _local = null;
    private ReentrantLock _linkMutex = new ReentrantLock();
    private boolean _isOpen = false;
    private ConnectType _connectType;
    private AtomicBoolean isClosing = new AtomicBoolean();
    private AtomicBoolean isOpening = new AtomicBoolean();

    /**
     * Constructor for connect to remote
     *
     * @param n
     * @param l
     */
    public TCPLink(TargetTuple t, OLDNetworkManager n, Layer l)
        throws IOException {
        super(n, l, true);

        InetAddress ia = InetAddress.getByName(n.getCurrentLocalHost());
        InetSocketAddress localBind = new InetSocketAddress(ia, 0);
        _socketChannel = SocketChannel.open();
        _socketChannel.configureBlocking(false);
        // non-blocking connect...
        _socketChannel.socket().bind(localBind);
        _socketChannel.connect(new InetSocketAddress(t.getIP(), t.getPort()));
        _local = (InetSocketAddress) _socketChannel.socket()
                                                   .getLocalSocketAddress();
        _targetTuple = t;
        setConnectionType(ConnectType.CONNECT_REUSE);
        // local helper function for the constructors
        init();
    }

    /**
     * Constructor for open after accept
     *
     * @param channel
     * @param n
     * @param l
     */
    public TCPLink(SocketChannel channel, OLDNetworkManager n, Layer l,
        boolean acceptRegister) throws IOException {
        super(n, l, true);
        _socketChannel = channel;
        _local = (InetSocketAddress) _socketChannel.socket()
                                                   .getLocalSocketAddress();
        _targetTuple = new TargetTuple(SipTransports.TCP_PROT,
                (InetSocketAddress) _socketChannel.socket()
                                                  .getRemoteSocketAddress());
        _socketChannel.configureBlocking(false);
        // local helper function for the constructors
        init();
        // listen on this link
        _socketChannel.register(_networkManager.getSelector(),
            SelectionKey.OP_READ, this);

        if (acceptRegister) {
            setConnectionType(ConnectType.ACCEPT_REGISTER);
            _networkManager.addAcceptedConnection(this);
            _networkManager.addActiveConnection(this);
            setConnectionType(ConnectType.CONNECT_REUSE);
        } else {
            setConnectionType(ConnectType.ACCEPT_UNREGISTER);
        }

        setOpen(true);
    }

    private ConnectType getConnectionType() {
        return _connectType;
    }

    private void setConnectionType(ConnectType type) {
        _connectType = type;
    }

    private void finishConnect() throws IOException {
        //
        // need to select for connecting
        //
        Selector connectSel = Selector.open();
        _socketChannel.register(connectSel, SelectionKey.OP_CONNECT);

        int sel;
        int selectRetries = _networkManager.getSipLinkTimeoutRetries();
        long waitTime = _networkManager.getSipLinkTimeout();

        try {
            while ((selectRetries > 0) &&
                    ((sel = connectSel.select(waitTime)) < 1)) {
                selectRetries--;

                if (sel < 0) {
                    //
                    // something really wrong !
                    //
                    _log.log(Level.SEVERE,
                        "connect error (select failed after " + selectRetries +
                        ". , socketChannel=" + _socketChannel.toString());
                    throw new IOException("connect error (select failed after " +
                        (_networkManager.getSipLinkTimeout() - selectRetries) +
                        " retries, socketChannel=" + _socketChannel.toString());
                }

                //
                // log that we were here, and for the n'th time
                //
                if (_log.isLoggable(Level.FINE)) {
                    _log.log(Level.FINE,
                        "Connect select retry (" +
                        (_networkManager.getSipLinkTimeoutRetries() -
                        selectRetries) + "), socketChannel=" +
                        _socketChannel.toString());
                }
            }
        } finally {
            connectSel.close();
        }

        if (selectRetries > 0) {
            _socketChannel.finishConnect();
        } else {
            throw new IOException("Unable to connect to " + _targetTuple);
        }
    }

    private void setOpen(boolean isOpen) {
        _isOpen = isOpen;
    }

    public synchronized boolean isOpen() {
        return (_socketChannel != null) ? _isOpen : false;
    }

    /**
     * Open this link to remote target. Should only be used together wih connect
     * constructor
     */
    public void open() throws IOException {
        if (getConnectionType() != ConnectType.CONNECT_REUSE) {
            throw new IOException(
                "Not allowed to open link, must use connect constructor " +
                getInfo());
        }

        if (_socketChannel == null) {
            throw new IOException("Connection is closed to " + getInfo());
        }

        // Ensure that only one thread is allowed to continue closing. The subsequent 
        // callers will return. 
        if (isOpening.getAndSet(true)) {
            return;
        }

        putTask(new OpenTask());
    }

    private void init() {
        try {
            int size = _socketChannel.socket().getReceiveBufferSize();

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

            _readBuffer = ByteBuffer.allocate(size);
        } catch (SocketException ignore) {
            _readBuffer = ByteBuffer.allocate(BUFFER_SIZE);
        }

        try {
            int size = _socketChannel.socket().getSendBufferSize();

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

            _writeBuffer = ByteBuffer.allocate(size);
        } catch (SocketException ignore) {
            _writeBuffer = ByteBuffer.allocate(BUFFER_SIZE);
        }
    }

    public TargetTuple getTargetTuple() {
        return _targetTuple;
    }

    public String getTransport() {
        return _targetTuple.getProtocol().name();
    }

    protected void closeImpl() {
        // Ensure that only one thread is allowed to continue closing. The subsequent 
        // callers will return. 
        if (isClosing.getAndSet(true)) {
            return;
        }

        _threadPool.execute(new Callable() {
                public Object call() throws Exception {
                    if (_socketChannel != null) {
                        _linkMutex.lock();

                        try {
                            if (_socketChannel != null) {
                                doClose();
                            } else {
                                if (_log.isLoggable(Level.INFO)) {
                                    _log.log(Level.INFO,
                                        "Tried to remove a closed link " +
                                        getInfo());
                                }
                            }
                        } finally {
                            _linkMutex.unlock();
                        }
                    }
                    return null;
                }
            });
    }

    private void doClose() {
        try {
            // Graceful shutdown
            if (!_socketChannel.socket().isOutputShutdown()) {
                try {
                    _socketChannel.socket().shutdownOutput();
                } catch (IOException e) {
                    if (_log.isLoggable(Level.INFO)) {
                        _log.log(Level.INFO,
                            "Failed to graceful shutdown link " + getInfo());
                    }
                }
            }

            _socketChannel.close();
        } catch (Throwable ignore) {
            if (_log.isLoggable(Level.INFO)) {
                _log.log(Level.INFO, "Failed to close link " + getInfo());
            }
        } finally {
            setOpen(false);
            _socketChannel = null;

            if (getConnectionType() == ConnectType.CONNECT_REUSE) {
                _networkManager.removeActiveConnection(this);
            } else if (getConnectionType() == ConnectType.ACCEPT_REGISTER) {
                _networkManager.removeAcceptedConnection(this);
            }
        }
    }

    public Object call() throws Exception{
        int n = -1;
        _readMutex.lock();

        try {
            if (_socketChannel != null) {
                n = _socketChannel.read(_readBuffer);

                if (_log.isLoggable(Level.FINE)) {
                    _log.log(Level.FINE, "Read == " + n);
                }
            } else {
                if (_log.isLoggable(Level.FINE)) {
                    _log.log(Level.FINE, "Link is closed " + getInfo());
                }
            }
        } catch (IOException ioe) {
            if (_log.isLoggable(Level.INFO)) {
                _log.log(Level.INFO,
                    "Got IOException for link: " + getInfo() + ", " + ioe +
                    ":" + ioe.getMessage());
            }
        } catch (Throwable t) {
            // FIXME alarm for e.g. OutOfMemoryError
            _log.log(Level.SEVERE,
                "Caught Throwable for socketChannel " + getTargetTuple() +
                ": ", t);
        } finally {
            // if n > 0 release of resources will be done by handleMessage
            if (n <= 0) {
                if (n < 0) {
                    closeImpl();
                }

                _networkManager.releaseSelectorSemaphore();
                _readMutex.unlock();
            }
        }

        if (n > 0) {
            _readBuffer.flip();
            _networkManager.releaseSelectorSemaphore();
            handleMessage(_readBuffer, getLocal(), SipTransports.TCP_PROT,
                getRemote());
        }
        return null;
    }

    public void write(InetSocketAddress remote, ByteBuffer message)
        throws IOException {
        if (_socketChannel == null) {
            throw new IOException("Link is closed " + getInfo());
        }

        while (message.hasRemaining()) {
            //
            // try to write first (succeeds most of the time)
            //
            if (_socketChannel.write(message) == 0) {
                //
                // couldn't write, need to select for writing
                //
                Selector writeSel = Selector.open();
                _socketChannel.register(writeSel, SelectionKey.OP_WRITE);

                int sel;
                int selectRetries = 0;
                long waitTime = _networkManager.getSipLinkTimeout();

                try {
                    while ((sel = writeSel.select(waitTime)) < 1) {
                        selectRetries++;

                        int left = message.remaining();

                        if (sel < 0) {
                            //
                            // something really wrong !
                            //
                            String info = _socketChannel.toString();
                            _log.log(Level.SEVERE,
                                "select failed (" + selectRetries +
                                "), socketChannel=" + info +
                                ", discarding buffer (" + left + " bytes) ...");
                            close();
                            message.clear();
                            throw new IOException(
                                "write error (select failed after " +
                                selectRetries + " retries, socketChannel=" +
                                info + ", discarding buffer (" + left +
                                " bytes).");
                        }

                        if (selectRetries == _networkManager.getSipLinkTimeoutRetries()) {
                            //
                            // max getEasLinkTimeoutSelect() retries
                            //
                            String info = _socketChannel.toString();
                            close();
                            message.clear();
                            throw new IOException("write error (max " +
                                _networkManager.getSipLinkTimeoutRetries() +
                                " select retries reached, socketChannel=" +
                                info + ", discarding buffer (" + left +
                                " bytes).");
                        }

                        //
                        // log that we were here, and for the n'th time
                        //
                        if (_log.isLoggable(Level.FINE)) {
                            _log.log(Level.FINE,
                                "selected (" + selectRetries +
                                "), socketChannel=" +
                                _socketChannel.toString() + ", buffer (" +
                                left + " bytes) ...");
                        }
                    }
                } finally {
                    writeSel.close();
                }

                //
                // Ok to write (select returned > 0)
                // Just go around the while loop again :-)
                //
            }
        }
    }

    public InetSocketAddress getLocal() {
        return _local;
    }

    public InetSocketAddress getRemote() {
        return _targetTuple.getSocketAddress();
    }

    public SelectableChannel getSelectableChannel() {
        return _socketChannel;
    }

    public void dispatch(final SipServletResponseImpl resp) {
        putTask(new QueuedTask() {
                public void doErrorAction(String cause) {
                }

                public String getError() {
                    return null;
                }

                public void run() {
                    // Redirect to find new link if channel is closed.
                    // 
                    // RFC3261 18.2.2 Sending Responses
                    // the response MUST be sent using the existing connection to the
                    // source of the original request that created the transaction, if that
                    // connection is still open...
                    // If that connection is no longer open, the server SHOULD open a
                    // connection to the IP address in the "received" parameter...
                    if (!isOpen()) {
                        if (getRemote()
                                    .equals(resp.getRemote().getSocketAddress())) {
                            // if the response "received" parameter saved in the getRemote()
                            // call is equal to this link, unfortunately the "received"
                            // parameter has already been established once and failed.
                            if (_log.isLoggable(Level.INFO)) {
                                _log.log(Level.INFO,
                                    "Unfortunately the 'received' parameter has already been established once and failed. Drop response " +
                                    resp.toString());
                            }

                            return;
                        }

                        if (_log.isLoggable(Level.INFO)) {
                            _log.log(Level.INFO,
                                "Try 'received' parameter since link is not open to " +
                                getInfo() + " for response " + resp.toString());
                        }

                        // try the "received" parameter...
                        resp.pushTransactionDispatcher(_networkManager);
                    }

                    TCPLink.super.dispatch(resp);
                }
            });
    }

    public String getInfo() {
        StringBuilder sb = new StringBuilder("[ip: ");
        sb.append(getTargetTuple().getIP());
        sb.append(":");
        sb.append(getTargetTuple().getPort());
        sb.append("/");
        sb.append(getTargetTuple().getProtocol().name());
        sb.append("]");

        return sb.toString();
    }

    /**
     * A task that opens a connection.
     * @author ejoelbi
     *
     */
    private final class OpenTask implements QueuedTask {
        /**
         * The rrror action: nothing
         */
        public void doErrorAction(String cause) {
        }

        /**
         * Gets a possible error: always null
         */
        public String getError() {
            return null;
        }

        /**
         * Do the connect.
         */
        public void run() {
            // Lock to ensure synchronization with closeImpl()
            _linkMutex.lock();

            try {
                // double checked-locking pattern
                if (_socketChannel == null) {
                    if (_log.isLoggable(Level.FINE)) {
                        _log.log(Level.FINE,
                            "Connection is closed to " + getInfo() +
                            " shutting down write processor for link");
                    }

                    shutdownWriteProcessor();
                }

                if (!_socketChannel.isConnected()) {
                    if (_socketChannel.isConnectionPending()) {
                        try {
                            finishConnect();
                            // local helper function for the constructors
                            init();
                            // need to register with read event
                            _networkManager.registerConnection(TCPLink.this);
                            // mark that this link is ready
                            setOpen(true);
                            // register with read event
                            _networkManager.getSelector().wakeup();
                        } catch (IOException e) {
                            doClose();

                            if (_log.isLoggable(Level.FINE)) {
                                _log.log(Level.FINE,
                                    "Failed to connect " + getInfo() +
                                    " exception: " + e +
                                    " shutting down write processor for link");
                            }

                            shutdownWriteProcessor();
                        } catch (Throwable t) {
                            doClose();

                            if (_log.isLoggable(Level.FINE)) {
                                _log.log(Level.FINE,
                                    "Failed to connect " + getInfo() +
                                    " exception: " + t +
                                    " shutting down write processor for link");
                            }

                            shutdownWriteProcessor();
                        }
                    } else {
                        if (_log.isLoggable(Level.INFO)) {
                            _log.log(Level.INFO,
                                "Lets close connection " + _targetTuple +
                                " since state is undefined [isConnected()=" +
                                _socketChannel.isConnected() +
                                ", isConnectionPending()=" +
                                _socketChannel.isConnectionPending() +
                                ", isOpen() " + _socketChannel.isOpen() +
                                ", isBlocking() " +
                                _socketChannel.isBlocking() +
                                ", isRegistered() " +
                                _socketChannel.isRegistered() + "]. ");
                        }

                        doClose();
                        shutdownWriteProcessor();
                    }
                }
            } finally {
                _linkMutex.unlock();
            }
        }
    }

    /**
     * Type of connection:
     * </p>
     * CONNECT_REUSE - Container has connected to remote and the link will be
     * saved in a list for re-use.
     * </p>
     * ACCEPT_REGISTER - Remote target has connected to container and an accept
     * has been issued. The link will be saved in a list and will be forced to
     * close on a UserCentricEvent.
     * </p>
     * ACCEPT_UNREGISTER - Remote target has connected to container and an accept
     * has been issued.
     */
    private enum ConnectType {CONNECT_REUSE,
        ACCEPT_REGISTER,
        ACCEPT_UNREGISTER;
    }
}
