/*_############################################################################
  _## 
  _##  SNMP4J - AbstractTransportMapping.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.SNMP4JSettings;
import org.snmp4j.TransportMapping;
import org.snmp4j.MessageDispatcher;

import java.io.IOException;

import org.snmp4j.TransportStateReference;
import org.snmp4j.log.LogAdapter;
import org.snmp4j.log.LogFactory;
import org.snmp4j.smi.Address;
import org.snmp4j.smi.OctetString;
import org.snmp4j.smi.TcpAddress;
import org.snmp4j.util.CommonTimer;
import org.snmp4j.util.DefaultThreadFactory;
import org.snmp4j.util.WorkerTask;

import java.nio.channels.SocketChannel;
import java.util.*;
import java.nio.ByteBuffer;
import java.util.concurrent.ConcurrentHashMap;

/**
 * The {@code AbstractTransportMapping} provides an abstract
 * implementation for the message dispatcher list and the maximum inbound
 * message size.
 *
 * @author Frank Fock
 * @version 3.12.0
 */
public abstract class AbstractTransportMapping<A extends Address> implements TransportMapping<A> {

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

    protected List<TransportListener> transportListener = new ArrayList<TransportListener>(1);
    protected transient volatile List<TransportAuditListener> transportAuditListeners;
    protected int maxInboundMessageSize = (1 << 16) - 1;
    protected boolean asyncMsgProcessingSupported = true;
    protected Set<A> suspendedAddresses = ConcurrentHashMap.newKeySet(5);
    protected WorkerTask listenWorkerTask;
    protected transient List<TransportStateListener> transportStateListeners;

    // 1 minute default timeout
    protected long connectionTimeout = 60000;
    protected CommonTimer socketCleaner;


    protected AbstractTransportMapping() {
    }

    public abstract Class<? extends Address> getSupportedAddressClass();

    public boolean isListening() {
        return (listenWorkerTask != null);
    }

    /**
     * Sends a message to the supplied address using this transport. If the target address has been suspended,
     * then instead actually sending the message on the wire, the method
     * {@link #handleDroppedMessageToSend(Address, byte[], TransportStateReference, long, int)} will be called.
     * To stop suspending of a target address, call {@link #resumeAddress(Address)} for that address.
     *
     * @param address
     *         an {@code Address} instance denoting the target address.
     * @param message
     *         the whole message as an array of bytes.
     * @param tmStateReference
     *         the (optional) transport model state reference as defined by
     *         RFC 5590 section 6.1.
     * @param timeoutMillis
     *         maximum number of milliseconds the connection creation might take (if connection-based).
     * @param maxRetries
     *         maximum retries during connection creation.
     *
     * @throws IOException
     *         if any underlying IO operation fails.
     */
    public abstract void sendMessage(A address, byte[] message,
                                     TransportStateReference tmStateReference, long timeoutMillis, int maxRetries)
            throws IOException;

    public synchronized void addTransportListener(TransportListener l) {
        if (!transportListener.contains(l)) {
            List<TransportListener> tlCopy = new ArrayList<>(transportListener);
            tlCopy.add(l);
            transportListener = tlCopy;
        }
    }

    public synchronized void removeTransportListener(TransportListener l) {
        if (transportListener != null && transportListener.contains(l)) {
            List<TransportListener> tlCopy = new ArrayList<>(transportListener);
            tlCopy.remove(l);
            transportListener = tlCopy;
        }
    }

    public synchronized void removeAllTransportListeners() {
        this.transportListener = new ArrayList<>(1);
    }

    /**
     * Adds a {@link TransportAuditListener} that is able to control (allow/deny) and audit the
     * communication of this transport mapping. If the same listener is already registered, it is not
     * added again.
     * @param l
     *    the {@link TransportAuditListener} to add.
     * @since 3.12.0
     */
    public synchronized void addTransportAuditListener(TransportAuditListener l) {
        List<TransportAuditListener> listeners = transportAuditListeners;
        if ((listeners == null) || !listeners.contains(l)) {
            List<TransportAuditListener> copy =
                    (listeners == null) ? new ArrayList<>(1) : new ArrayList<>(listeners);
            copy.add(l);
            transportAuditListeners = copy;
        }
    }

    /**
     * Removes a previously added {@link TransportAuditListener}.
     * @param l
     *    the {@link TransportAuditListener} to remove.
     * @since 3.12.0
     */
    public synchronized void removeTransportAuditListener(TransportAuditListener l) {
        List<TransportAuditListener> listeners = transportAuditListeners;
        if ((listeners != null) && listeners.contains(l)) {
            List<TransportAuditListener> copy = new ArrayList<>(listeners);
            copy.remove(l);
            transportAuditListeners = copy.isEmpty() ? null : copy;
        }
    }

    /**
     * Removes all registered {@link TransportAuditListener}s.
     * @since 3.12.0
     */
    public synchronized void removeAllTransportAuditListeners() {
        this.transportAuditListeners = null;
    }

    /**
     * Checks whether at least one {@link TransportAuditListener} is registered with this transport
     * mapping. This method can be used to quickly check whether building and firing a
     * {@link TransportAuditEvent} is necessary at all, in order to avoid the associated overhead when
     * no audit listener is registered.
     * @return
     *    {@code true} if at least one {@link TransportAuditListener} is registered,
     *    {@code false} otherwise.
     * @since 3.12.0
     */
    public boolean isTransportAudit() {
        List<TransportAuditListener> listeners = transportAuditListeners;
        return (listeners != null) && !listeners.isEmpty();
    }

    /**
     * Returns the listen address of this transport mapping for use in {@link TransportAuditEvent}s
     * without propagating exceptions. Some transport mappings throw an exception from
     * {@link #getListenAddress()} when they are not listening yet; this helper returns {@code null}
     * in that case so that audit event creation never fails.
     * @return
     *    the listen address or {@code null} if it cannot be determined.
     * @since 3.12.0
     */
    protected A getAuditListenAddress() {
        try {
            return getListenAddress();
        } catch (RuntimeException ex) {
            return null;
        }
    }

    /**
     * Notifies all registered {@link TransportAuditListener}s about the supplied
     * {@link TransportAuditEvent}. All listeners are always notified (so that they can audit the
     * event), even if one of them denies the operation.
     * @param event
     *    the {@link TransportAuditEvent} to fire.
     * @return
     *    {@code true} if the operation described by {@code event} is permitted (i.e. no listener
     *    denied it or no listener is registered at all), {@code false} if at least one listener
     *    denied the operation. For {@link TransportAuditEvent.TransportOperation#CLOSE} events the
     *    result should be ignored by the caller.
     * @since 3.12.0
     */
    protected boolean fireTransportAuditEvent(TransportAuditEvent event) {
        final List<TransportAuditListener> listeners = transportAuditListeners;
        if ((listeners == null) || listeners.isEmpty()) {
            return true;
        }
        boolean permitted = true;
        for (TransportAuditListener listener : listeners) {
            try {
                if (!listener.transportAudit(event)) {
                    permitted = false;
                }
            } catch (RuntimeException ex) {
                logger.error("Exception in transport audit listener: " + ex.getMessage(), ex);
                if (SNMP4JSettings.isForwardRuntimeExceptions()) {
                    throw ex;
                }
            }
        }
        return permitted;
    }

    /**
     * Processes an incoming message received from the specified address and notifies
     * the registered transport listeners. If transport auditing is enabled, an audit event
     * will be fired before processing the message.
     *
     * @param address the address of the sender from which the message is received
     * @param buf the {@code ByteBuffer} containing the received message
     * @param tmStateReference a reference to the transport state associated with this message
     */
    protected void fireProcessMessage(A address, ByteBuffer buf,
                                      TransportStateReference tmStateReference) {
        if (isTransportAudit()) {
            byte[] message = new byte[buf.remaining()];
            buf.duplicate().get(message);
            if (!fireTransportAuditEvent(new TransportAuditEvent(this,
                    TransportAuditEvent.TransportOperation.RECEIVE, address, getAuditListenAddress(), message))) {
                if (logger.isDebugEnabled()) {
                    logger.debug("Transport audit denied processing of message received from " + address);
                }
                return;
            }
        }
        if (transportListener != null) {
            for (TransportListener aTransportListener : transportListener) {
                aTransportListener.processMessage(this, address, buf, tmStateReference);
            }
        }
    }

    /**
     * Closes this resource and releases any associated system resources.
     * If the resource is already closed, invoking this method has no effect.
     *
     * @throws IOException if an I/O error occurs while closing the resource
     */
    public abstract void close() throws IOException;

    /**
     * Suspend sending of messages to the specified address, regardless if a connection is already established or
     * not. To be able to send messages again to the specified address using
     * {@link #sendMessage(Address, byte[], TransportStateReference, long, int)}, call
     * {@link #resumeAddress(Address)}.
     * @param addressToSuspendSending
     *    an arbitrary remote address for which any messages send by
     *    {@link #sendMessage(Address, byte[], TransportStateReference, long, int)} should be dropped before sending
     *    and reopening a connection to that address.
     * @since 3.4.4
     */
    public void suspendAddress(A addressToSuspendSending) {
        if (suspendedAddresses.add(addressToSuspendSending)) {
            logger.info("Messages sending to "+addressToSuspendSending+" suspended");
        }
    }

    /**
     * Resume sending of messages to the specified address.
     * @param addressToResumeSending
     *    an arbitrary remote address for which any messages send by
     *    {@link #sendMessage(Address, byte[], TransportStateReference, long, int)} should be dropped before sending
     *    and reopening a connection to that address.
     * @return
     *    {@code true} if the specified address was previously suspended and is now resumed to allow sending messages,
     *    {@code false} otherwise.
     * @since 3.4.4
     */
    public boolean resumeAddress(A addressToResumeSending) {
        boolean resumed = suspendedAddresses.remove(addressToResumeSending);
        if (resumed) {
            logger.info("Messages sending to "+addressToResumeSending+" resumed");
        }
        return resumed;
    }

    /**
     * Handle a message that could not be send to the specified address, because there is no server socket for
     * receiving responses.
     * @param address
     *         an {@code Address} instance denoting the target address.
     * @param message
     *         the whole message as an array of bytes.
     * @param transportStateReference
     *         the (optional) transport model state reference as defined by
     *         RFC 5590 section 6.1.
     * @param timeoutMillis
     *         maximum number of milliseconds the connection creation might take (if connection based).
     * @param maxRetries
     *         maximum retries during connection creation.
     * @since 3.4.4
     */
    protected void handleDroppedMessageToSend(A address, byte[] message,
                                              TransportStateReference transportStateReference,
                                              long timeoutMillis, int maxRetries) {
        logger.warn("Dropped message, because this transport mapping is suspended: address="+
                address+", message="+ OctetString.fromByteArray(message).toHexString());
    }

    /**
     * Starts listening for incoming connections or events. This method is intended to be
     * implemented by subclasses to define the specific behavior required for listening
     * operations in various contexts (e.g., network connections, event streams).
     *
     * @throws IOException if an I/O error occurs during the listening process.
     */
    public abstract void listen() throws IOException;

    /**
     * Gets the {@link CommonTimer} that controls socket cleanup operations.
     *
     * @return a socket cleaner timer.
     * @since 3.0
     */
    public CommonTimer getSocketCleaner() {
        return socketCleaner;
    }

    public int getMaxInboundMessageSize() {
        return maxInboundMessageSize;
    }

    /**
     * Returns {@code true} if asynchronous (multi-threaded) message
     * processing may be implemented. The default is {@code true}.
     *
     * @return if {@code false} is returned the
     * {@link MessageDispatcher#processMessage(org.snmp4j.TransportMapping, org.snmp4j.smi.Address, java.nio.ByteBuffer, org.snmp4j.TransportStateReference)}
     * method must not return before the message has been entirely processed.
     */
    public boolean isAsyncMsgProcessingSupported() {
        return asyncMsgProcessingSupported;
    }

    /**
     * Specifies whether this transport mapping has to support asynchronous
     * messages processing or not.
     *
     * @param asyncMsgProcessingSupported
     *         if {@code false} the {@link MessageDispatcher#processMessage(org.snmp4j.TransportMapping, org.snmp4j.smi.Address, java.nio.ByteBuffer, org.snmp4j.TransportStateReference)}
     *         method must not return before the message has been entirely processed,
     *         because the incoming message buffer is not copied before the message
     *         is being processed. If {@code true} the message buffer is copied
     *         for each call, so that the message processing can be implemented
     *         asynchronously.
     */
    public void setAsyncMsgProcessingSupported(
            boolean asyncMsgProcessingSupported) {
        this.asyncMsgProcessingSupported = asyncMsgProcessingSupported;
    }

    /**
     * Changes the priority of the listen thread for this UDP transport mapping.
     * This method has no effect, if called before {@link #listen()} has been
     * called for this transport mapping.
     * @param newPriority
     *         the new priority.
     * @see Thread#setPriority(int)
     * @since 3.6.0
     */
    public void setPriority(int newPriority) {
        WorkerTask lt = getListenWorkerTask();
        if (lt instanceof Thread) {
            ((Thread) lt).setPriority(newPriority);
        }
        else if (lt instanceof DefaultThreadFactory.WorkerThread) {
            ((DefaultThreadFactory.WorkerThread) lt).getThread().setPriority(newPriority);
        }
    }

    /**
     * Returns the priority of the internal listen thread.
     *
     * @return a value between {@link Thread#MIN_PRIORITY} and
     * {@link Thread#MAX_PRIORITY}.
     * @since 3.6.0
     */
    public int getPriority() {
        WorkerTask lt = getListenWorkerTask();
        if (lt instanceof Thread) {
            return ((Thread) lt).getPriority();
        }
        else if (lt instanceof DefaultThreadFactory.WorkerThread) {
            return ((DefaultThreadFactory.WorkerThread) lt).getThread().getPriority();
        } else {
            return Thread.NORM_PRIORITY;
        }
    }

    /**
     * Sets the name of the listen thread for this UDP transport mapping.
     * This method has no effect, if called before {@link #listen()} has been
     * called for this transport mapping.
     * @param name
     *         the new thread name.
     * @since 3.6.0
     */
    public void setThreadName(String name) {
        WorkerTask lt = getListenWorkerTask();
        if (lt instanceof Thread) {
            ((Thread) lt).setName(name);
        }
        else if (lt instanceof DefaultThreadFactory.WorkerThread) {
            ((DefaultThreadFactory.WorkerThread) lt).getThread().setName(name);
        }
    }

    /**
     * Returns the name of the listen thread.
     *
     * @return the thread name if in listening mode, otherwise {@code null}.
     * @since 3.6.0
     */
    public String getThreadName() {
        WorkerTask lt = getListenWorkerTask();
        if (lt instanceof Thread) {
            return ((Thread) lt).getName();
        }
        else if (lt instanceof DefaultThreadFactory.WorkerThread) {
            return ((DefaultThreadFactory.WorkerThread) lt).getThread().getName();
        } else {
            return null;
        }
    }

    /**
     * Add a {@link TransportStateListener} to get {@link TransportStateEvent}s if the state of this transport mapping
     * changes.
     * @param l
     *    the listener callback interface.
     * @since 3.7.0
     */
    public synchronized void addTransportStateListener(TransportStateListener l) {
        if (transportStateListeners == null) {
            transportStateListeners = new ArrayList<>(2);
        }
        transportStateListeners.add(l);
    }

    /**
     * Remove a {@link TransportStateListener} from this transport mapping.
     * @param l
     *    the listener callback interface to be removed.
     * @since 3.7.0
     */
    public synchronized void removeTransportStateListener(TransportStateListener l) {
        if (transportStateListeners != null) {
            transportStateListeners.remove(l);
        }
    }

    public abstract A getListenAddress();

    protected void fireConnectionStateChanged(TransportStateEvent change) {
        if (logger.isDebugEnabled()) {
            logger.debug("Firing transport state event: " + change);
        }
        final List<TransportStateListener> listenersFinalRef = transportStateListeners;
        if (listenersFinalRef != null) {
            try {
                List<TransportStateListener> listeners;
                synchronized (listenersFinalRef) {
                    listeners = new ArrayList<TransportStateListener>(listenersFinalRef);
                }
                for (TransportStateListener listener : listeners) {
                    listener.connectionStateChanged(change);
                }
            } catch (RuntimeException ex) {
                logger.error("Exception in fireConnectionStateChanged: " + ex.getMessage(), ex);
                if (SNMP4JSettings.isForwardRuntimeExceptions()) {
                    throw ex;
                }
            }
        }
    }

    /**
     * Gets the {@link WorkerTask} that is responsible for receiving new messages.
     * @return
     *    a {@link WorkerTask} instance which is most likely a {@link org.snmp4j.util.DefaultThreadFactory.WorkerThread}.
     * @since 3.7.0
     */
    public WorkerTask getListenWorkerTask() {
        return listenWorkerTask;
    }

    /**
     * Gets an unmodifiable set of the suspended addresses of this
     * @return
     *    an unmodifiable set of the suspended addresses.
     */
    public Set<A> getSuspendedAddresses() {
        return Collections.unmodifiableSet(suspendedAddresses);
    }
}
