/*_############################################################################
  _## 
  _##  SNMP4J - AbstractConnectionOrientedTransportMapping.java  
  _## 
  _##  Copyright (C) 2003-2026  Frank Fock (SNMP4J.org)
  _##  
  _##  Licensed under the Apache License, Version 2.0 (the "License");
  _##  you may not use this file except in compliance with the License.
  _##  You may obtain a copy of the License at
  _##  
  _##      http://www.apache.org/licenses/LICENSE-2.0
  _##  
  _##  Unless required by applicable law or agreed to in writing, software
  _##  distributed under the License is distributed on an "AS IS" BASIS,
  _##  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  _##  See the License for the specific language governing permissions and
  _##  limitations under the License.
  _##  
  _##########################################################################*/

package org.snmp4j.transport;

import org.snmp4j.log.LogAdapter;
import org.snmp4j.log.LogFactory;
import org.snmp4j.smi.Address;
import org.snmp4j.smi.TcpAddress;
import org.snmp4j.util.WorkerTask;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketOption;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * The {@link AbstractConnectionOrientedTransportMapping} implements generic functions for a connection
 * oriented transport mapping for server and client connections.
 *
 * @param <A> the address type.
 * @param <S> the socket entry type.
 * @author Frank Fock
 * @since 3.7.0
 */
public abstract class AbstractConnectionOrientedTransportMapping<A extends Address, S extends AbstractSocketEntry<A>>
        extends AbstractTransportMapping<A> implements ConnectionOrientedTransportMapping<A> {

    private static final LogAdapter logger = LogFactory.getLogger(AbstractConnectionOrientedTransportMapping.class);

    protected boolean serverEnabled = false;
    protected Map<A, S> sockets = new ConcurrentHashMap<>();
    private int maxBusyLoops = DefaultTcpTransportMapping.DEFAULT_MAX_BUSY_LOOPS;

    protected long gracefulShutdownTimeoutMillis = 0;
    protected long connectionTimeout = 0;

    protected AbstractConnectionOrientedTransportMapping() {
    }

    protected synchronized void timeoutSocket(AbstractSocketEntry<A> entry) {
        if ((connectionTimeout > 0) && (getSocketCleaner() != null)) {
            SocketTimeout<A> socketTimeout = new SocketTimeout<A>(this, entry);
            entry.setSocketTimeout(socketTimeout);
            getSocketCleaner().schedule(socketTimeout, connectionTimeout);
        }
    }

    /**
     * Gets the graceful shutdown timeout for the server socket in milliseconds.
     * @return
     *    the timeout in milliseconds or 0 if no timeout is set.
     */
    public long getGracefulShutdownTimeoutMillis() {
        return gracefulShutdownTimeoutMillis;
    }

    /**
     * Sets the graceful shutdown timeout for the server socket in milliseconds.
     * A value of 0 disables the timeout (default).
     * If the timeout is exceeded, the server will forcefully close the server socket.
     * @param gracefulShutdownTimeoutMillis
     *    the timeout in milliseconds.
     */
    public void setGracefulShutdownTimeoutMillis(long gracefulShutdownTimeoutMillis) {
        if (gracefulShutdownTimeoutMillis < 0) {
            throw new IllegalArgumentException("Timeout must be >= 0");
        }
        this.gracefulShutdownTimeoutMillis = gracefulShutdownTimeoutMillis;
    }

    /**
     * Gets the connection timeout. This timeout specifies the time a connection
     * may be idle before it is closed.
     *
     * @return long
     * the idle timeout in milliseconds.
     */
    public long getConnectionTimeout() {
        return connectionTimeout;
    }

    /**
     * Sets the connection timeout. This timeout specifies the time a connection
     * may be idle before it is closed.
     *
     * @param connectionTimeout
     *         the idle timeout in milliseconds. A zero or negative value will disable
     *         any timeout and connections opened by this transport mapping will stay
     *         opened until they are explicitly closed.
     */
    public void setConnectionTimeout(long connectionTimeout) {
        this.connectionTimeout = connectionTimeout;
    }

    /**
     * Sets optional server socket options. The default implementation does
     * nothing.
     *
     * @param serverSocket
     *         the {@link ServerSocket} to apply additional non-default options.
     */
    protected void setSocketOptions(ServerSocket serverSocket) {
    }

    /**
     * Sets the value of a specific socket option.
     * <p>
     * This method allows configuring transport-specific socket options, which may
     * impact the behavior of the underlying socket implementation. The specific
     * options and the corresponding values depend on the platform and socket
     * implementation.
     *
     * @param <T>
     *        the type of the value associated with the socket option
     * @param socketOption
     *        the {@link SocketOption} to set; cannot be null
     * @param value
     *        the value to be set for the specified socket option; may be null if the
     *        option supports null values
     * @throws UnsupportedOperationException
     *         if the specified option is not supported by the underlying socket
     *         implementation
     * @throws IllegalArgumentException
     *         if the specified value is not valid for the given socket option
     * @throws IOException
     *         if an I/O error occurs while attempting to set the socket option
     * @since 3.10.0
     */
    protected abstract <T> void setSocketOption(SocketOption<T> socketOption, T value) throws IOException;

    @Override
    public boolean isServerEnabled() {
        return serverEnabled;
    }

    @Override
    public void setServerEnabled(boolean serverEnabled) {
        this.serverEnabled = serverEnabled;
    }

    protected void closeSockets(Map<A, S> sockets) {
        List<A> closedSockets = new LinkedList<>();
        for (S entry : sockets.values()) {
            SocketChannel sc = entry.getSocketChannel();
            if (sc != null) {
                try {
                    try {
                        sc.socket().close();
                    } catch (UnsupportedOperationException unsupportedOperationException) {
                        // unix domain socket
                        sc.close();
                    }
                    if (logger.isInfoEnabled()) {
                        logger.info("Socket to " + entry.getPeerAddress() + " closed");
                    }
                    TransportStateEvent e =
                            new TransportStateEvent(this, entry.getPeerAddress(),
                                    TransportStateEvent.STATE_CLOSED, null);
                    fireConnectionStateChanged(e);
                    closedSockets.add(entry.getPeerAddress());

                } catch (IOException iox) {
                    // ignore
                    logger.debug(iox);
                }
            }
        }
        for (A remoteAddress : closedSockets) {
            sockets.remove(remoteAddress);
        }
    }

    /**
     * Closes a connection to the supplied remote address if it is open forcefully (by TCP RST). This
     * method is particularly useful when not using a timeout for remote
     * connections.
     *
     * @param remoteAddress
     *         the address of the peer socket.
     *
     * @return {@code true} if the connection has been closed and
     * {@code false} if there was nothing to close.
     * @throws IOException
     *         if the remote address cannot be closed due to an IO exception.
     * @since 1.7.1
     */
    public synchronized boolean close(A remoteAddress) throws IOException {
       return close(remoteAddress, true);
    }

    /**
     * Closes a connection to the supplied remote address if it is open. This
     * method is particularly useful when not using a timeout for remote
     * connections.
     *
     * @param remoteAddress
     *         the address of the peer socket.
     * @param force
     *         if true, the connection will be forcefully closed with TCP RST. If false, a TCP FIN message is sent
     *         on the outbound connection only.
     * @return {@code true} if the connection has been closed and
     * {@code false} if there was nothing to close.
     * @throws IOException
     *         if the remote address cannot be closed due to an IO exception.
     * @since 3.10.0
     */
    public synchronized boolean close(A remoteAddress, boolean force) throws IOException {
        if (logger.isDebugEnabled()) {
            logger.debug("Closing socket for peer address " + remoteAddress);
        }
        if (isTransportAudit()) {
            fireTransportAuditEvent(new TransportAuditEvent(this,
                    TransportAuditEvent.TransportOperation.CLOSE, remoteAddress, getAuditListenAddress(), null));
        }
        AbstractSocketEntry<A> entry = sockets.remove(remoteAddress);
        if (entry != null) {
            if (entry.getSocketTimeout() != null) {
                entry.getSocketTimeout().cancel();
            }
            SocketChannel sc = entry.getSocketChannel();
            if (sc != null) {
                try {
                    if (force) {
                        sc.socket().close();
                    }
                    else {
                        sc.shutdownOutput();
                    }
                } catch (UnsupportedOperationException unsupportedOperationException) {
                    if (force) {
                        sc.close();
                    }
                    else {
                        sc.shutdownOutput();
                    }
                    if (logger.isDebugEnabled()) {
                        logger.debug("Closed socket channel for peer address " + remoteAddress);
                    }
                }
                if (logger.isInfoEnabled()) {
                    logger.info("Socket to " + entry.getPeerAddress() + " closed "+(force?"(forced)":""));
                }
                TransportStateEvent e =
                        new TransportStateEvent(this, entry.getPeerAddress(),
                                TransportStateEvent.STATE_CLOSED, null);
                fireConnectionStateChanged(e);
            }
            return true;
        }
        return false;
    }

    /**
     * Closes all open sockets and stops the internal server thread that
     * processes messages.
     */
    public synchronized void close() throws IOException {
        if (isTransportAudit()) {
            fireTransportAuditEvent(new TransportAuditEvent(this,
                    TransportAuditEvent.TransportOperation.CLOSE, null, getAuditListenAddress(), null));
        }
        WorkerTask st = listenWorkerTask;
        listenWorkerTask = null;
        if (st != null) {
            st.shutdown();
            for (S entry : sockets.values()) {
                entry.closeSession();
                TransportStateEvent e =
                        new TransportStateEvent(this, entry.getPeerAddress(), TransportStateEvent.STATE_CLOSED,
                                null);
                fireConnectionStateChanged(e);
            }
            try {
                if (gracefulShutdownTimeoutMillis > 0) {
                    if (!st.join(gracefulShutdownTimeoutMillis)) {
                        st.terminate();
                        st.interrupt();
                    }
                }
                else {
                    st.terminate();
                    st.interrupt();
                    st.join();
                }
            } catch (InterruptedException ex) {
                logger.warn(ex);
            }
            closeSockets(sockets);
            sockets.clear();
            if (getSocketCleaner() != null) {
                getSocketCleaner().cancel();
            }
            socketCleaner = null;
        }
    }

    public abstract void wakeupServerSelector();

    /**
     * Gets an unmodifiable map of the {@link AbstractSocketEntry} instances associated with this transport mapping.
     * @return
     *    an unmodifiable map from {@link Address} to {@link AbstractSocketEntry}.
     * @since 3.7.0
     */
    public Map<A, S> getSockets() {
        return Collections.unmodifiableMap(sockets);
    }

    protected void cancelNonServerSelectionKey(SelectionKey sk) {
        if (!sk.isAcceptable()) {
            sk.cancel();
        }
    }

    protected int getMaxBusyLoops() {
        return maxBusyLoops;
    }

    protected void setMaxBusyLoops(int maxBusyLoops) {
        this.maxBusyLoops = maxBusyLoops;
    }
}
