/*_############################################################################
  _## 
  _##  SNMP4J-AgentX - AgentXQueue.java  
  _## 
  _##  Copyright (C) 2005-2026  Frank Fock (SNMP4J.org)
  _##  
  _##  This program is free software; you can redistribute it and/or modify
  _##  it under the terms of the GNU General Public License version 2 as 
  _##  published by the Free Software Foundation.
  _##
  _##  This program is distributed in the hope that it will be useful,
  _##  but WITHOUT ANY WARRANTY; without even the implied warranty of
  _##  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  _##  GNU General Public License for more details.
  _##
  _##  You should have received a copy of the GNU General Public License
  _##  along with this program; if not, write to the Free Software
  _##  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, 
  _##  MA  02110-1301  USA
  _##  
  _##########################################################################*/


package org.snmp4j.agent.agentx.master;

import java.io.Serial;
import java.io.Serializable;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.function.Function;

import org.snmp4j.PDU;
import org.snmp4j.agent.MOServer;
import org.snmp4j.agent.agentx.AgentXProtocol;
import org.snmp4j.agent.request.SnmpRequest;
import org.snmp4j.log.LogAdapter;
import org.snmp4j.log.LogFactory;
import org.snmp4j.smi.Address;
import org.snmp4j.smi.OID;
import org.snmp4j.smi.VariableBinding;
import org.snmp4j.agent.DefaultMOScope;
import org.snmp4j.agent.MOScope;
import org.snmp4j.smi.OctetString;

/**
 * The {@link AgentXQueue} holds AgentX requests that are processed by the master agent by sending the requests
 * to the subagents and waiting for responses.
 */
public class AgentXQueue implements Serializable {

    private static final LogAdapter LOGGER = LogFactory.getLogger(AgentXQueue.class);
    @Serial
    private static final long serialVersionUID = 4177433540588469413L;

    /**
     * The ordered list of AgentX requests waiting to be processed.
     */
    private final LinkedList<AgentXQueueEntry<?>> queue = new LinkedList<>();
    /**
     * The managed object servers used for GETBULK optimization.
     */
    private transient MOServer[] servers;

    /**
     * Creates a new {@link AgentXQueue}.
     */
    public AgentXQueue() {
    }

    /**
     * Sets the {@link MOServer} instances for GETBULK optimization.
     * @param servers
     *    an array of {@link MOServer} instances.
     */
    public void setServer4BulkOptimization(MOServer[] servers) {
        this.servers = servers;
    }

    /**
     * Gets the {@link MOServer} instances used for GETBULK optimization.
     * @return
     *    {@link MOServer} instances for GETBULK optimization.
     */
    public MOServer[] getServer4BulkOptimization() {
        return this.servers;
    }

    /**
     * Adds a variable binding to a pending SET request (if not in progress yet) or creating a new SET request.
     * @param vb
     *    the variable binding.
     * @param subRequest
     *    the SNMP sub-request
     * @param entry
     *    the registration entry on whose behalf this request is being processed.
     * @param <A>
     *     address type.
     * @return
     *     {@code true} if the request has been added.
     */
    public synchronized <A extends Address> boolean add(VariableBinding vb, SnmpRequest.SnmpSubRequest subRequest,
                                    AgentXRegEntry<A> entry) {
        SnmpRequest request = subRequest.getRequest();
        @SuppressWarnings("unchecked")
        AgentXPendingSet<A> pending = (AgentXPendingSet<A>) get(entry.getSession().getSessionID(),
                request.getTransactionID());
        if (pending == null) {
            pending = new AgentXPendingSet<A>(entry, subRequest.getSnmpRequest());
            insertIntoQueue(request.getTransactionID(), pending);
        }
        if (!pending.isPending()) {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Variable binding " + vb +
                        " not added because AgentX request " + pending +
                        " is waiting for response");
            }
            return false;
        }
        pending.add(subRequest, vb);
        return true;
    }

    private synchronized <A extends Address> void insertIntoQueue(int transactionID, AgentXPending<A> pending) {
        AgentXRegEntry<A> reg = pending.getRegistration();
        int timeout = reg.getTimeout();
        if (timeout == 0) {
            timeout = reg.getSession().getTimeout() & 0xFF;
        }
        pending.setTimeout(timeout);

        @SuppressWarnings("unchecked")
        AgentXQueueEntry<A> entry = (AgentXQueueEntry<A>) getQueueEntry(transactionID, false);
        if (entry == null) {
            entry = new AgentXQueueEntry<A>(transactionID);
            queue.add(entry);
        }
        entry.addEntry(pending);
    }

    /**
     * Adds a search range to a GET request.
     *
     * @param <A>         address type.
     * @param searchRange the search range to add.
     * @param entry       the registration entry.
     * @param repeater    {@code true} if the added search Range is a repeating range.
     * @return {@code true} if the request has been added.
     */
    public synchronized <A extends Address> boolean add(AgentXSearchRange searchRange, AgentXRegEntry<A> entry,
                                                        boolean repeater) {
        return add(searchRange, entry, repeater, null);
    }

    /**
     * Adds a search range to a GET request.
     *
     * @param <A>         address type.
     * @param searchRange the search range to add.
     * @param entry       the registration entry.
     * @param repeater    {@code true} if the added search Range is a repeating range.
     * @param filter      an optional filter that accepts only certain OIDs returned by a subagent, for example used
     *                    by VACM or for skipping known implementation errors in subagents.
     *                    See {@link AgentXPendingGet#oidFilter}.
     * @since 4.1.0
     * @return {@code true} if the request has been added.
     */
    public synchronized <A extends Address> boolean add(AgentXSearchRange searchRange, AgentXRegEntry<A> entry,
                                                        boolean repeater, Function<OID, Boolean> filter) {
        SnmpRequest request = searchRange.getReferenceSubRequest().getRequest();
        @SuppressWarnings("unchecked")
        AgentXPendingGet<A> pending = (AgentXPendingGet<A>) get(entry.getSession().getSessionID(),
                request.getTransactionID());
        if (pending == null) {
            // optimize upper bound if server is set
            if ((servers != null) && (request.getSource().getPDU().getType() == PDU.GETBULK)) {
                optimizeSearchRange(searchRange, entry);
            }
            pending = new AgentXPendingGet<>(entry, request, searchRange, filter);
            insertIntoQueue(request.getTransactionID(), pending);
        } else if (pending.isPending()) {
            // do nothing special
            if (request.getSource().getPDU().getType() == PDU.GETBULK) {
                for (AgentXSearchRange psr : pending.getSearchRanges()) {
                    int repCount = request.getRepeaterCount();
                    if ((repCount > 0) &&
                            (((searchRange.getReferenceSubRequest().getIndex() - pending.getNonRepeater()) -
                                    (psr.getReferenceSubRequest().getIndex() - request.getNonRepeaters())) %
                                    repCount == 0)) {
                        // this is a new repetition -> ignore it this time and send out
                        // AgentX request
                        if (LOGGER.isDebugEnabled()) {
                            LOGGER.debug("Repetition not added because of pending AgentX" +
                                    " processing of " + pending + " and repetition " + psr);
                        }
                        return false;
                    }
                }
                // optimize upper bound if server is set
                if (servers != null) {
                    optimizeSearchRange(searchRange, entry);
                }
            }
            pending.addSearchRange(searchRange);
        } else {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Search range " + searchRange +
                        " not added because AgentX request " + pending +
                        " is not pending");
            }
            return false;
        }
        if (!repeater) {
            pending.incNonRepeater();
        }
        return true;
    }

    private MOServer getServer(OctetString context) {
        MOServer[] sc = servers;
        for (MOServer s : sc) {
            if (s.isContextSupported(context)) {
                return s;
            }
        }
        return null;
    }

    /**
     * Optimizes the search range to limit sub-agent request to upper bounds know in the master apriori.
     * @param searchRange
     *    the search range to optimize.
     * @param entry
     *    the registration entry for the search.
     */
    protected void optimizeSearchRange(AgentXSearchRange searchRange, AgentXRegEntry<?> entry) {
        DefaultMOScope scope = new DefaultMOScope(searchRange.getUpperBound(),
                !searchRange.isUpperIncluded(),
                null, false);
        AgentXNodeQuery query =
                new AgentXNodeQuery(entry.getContext(), scope, AgentXNodeQuery.QUERY_ALL);
        MOScope requestScope = searchRange.getReferenceSubRequest().getScope();
        MOServer server = getServer(entry.getContext());
        if (server == null) {
            // nothing to do then
            return;
        }
        for (AgentXNode node = server.lookup(query, AgentXNode.class);
             node != null;
             node = server.lookup(nextQuery(query, node), AgentXNode.class)) {
            AgentXRegEntry<?> activeReg = node.getActiveRegistration();
            MOScope region = node.getScope();
            if ((activeReg != null) && (activeReg.getSession().equals(entry.getSession()))) {
                if ((requestScope.getUpperBound() != null) &&
                        (requestScope.getUpperBound().compareTo(region.getUpperBound()) <= 0)) {
                    searchRange.setUpperBound(requestScope.getUpperBound());
                    searchRange.setUpperIncluded(requestScope.isUpperIncluded());
                    break;
                }
                searchRange.setUpperBound(region.getUpperBound());
                searchRange.setUpperIncluded(region.isUpperIncluded());
            } else {
                if ((searchRange.getUpperBound() == null) ||
                        (searchRange.getUpperBound().compareTo(region.getLowerBound()) >= 0)) {
                    searchRange.setUpperBound(region.getLowerBound());
                    searchRange.setUpperIncluded(!region.isLowerIncluded());
                }
                break;
            }
        }
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Optimized upper bound for bulk AgentX request to " +
                    searchRange);
        }
    }

    private static AgentXNodeQuery nextQuery(AgentXNodeQuery lastQuery, AgentXNode lastNode) {
        if (lastNode != null) {
            lastQuery.getMutableScope().setLowerBound(
                    lastNode.getScope().getUpperBound());
            lastQuery.getMutableScope().setLowerIncluded(false);
        }
        return lastQuery;
    }

    /**
     * Returns the AgentX request in the queue identified by an AgentX session ID
     * and a transaction ID.
     *
     * @param sessionID
     *         the session ID.
     * @param transactionID
     *         the transaction ID.
     *
     * @return the associated {@code AgentXPending} instance or {@code null}
     * if no such request exists.
     */
    public synchronized AgentXPending<?> get(int sessionID, int transactionID) {
        AgentXQueueEntry<?> entry = getQueueEntry(transactionID, false);
        if (entry != null) {
            return entry.get(sessionID, false);
        }
        return null;
    }

    /**
     * Returns the AgentX request in the queue identified by an AgentX session ID
     * and a transaction ID and removes that request from the queue.
     *
     * @param sessionID
     *         the session ID.
     * @param transactionID
     *         the transaction ID.
     *
     * @return the associated {@code AgentXPending} instance or {@code null}
     * if no such request exists.
     */
    public synchronized AgentXPending<?> remove(int sessionID, int transactionID) {
        AgentXQueueEntry<?> entry = getQueueEntry(transactionID, false);
        if (entry != null) {
            return entry.get(sessionID, true);
        }
        return null;
    }


    /**
     * Return all pending AgentX requests for the specified transaction ID.
     *
     * @param transactionID
     *         a transcation ID.
     *
     * @return a possibly empty List of pending requests.
     */
    public synchronized AgentXQueueEntry<?> get(int transactionID) {
        return getQueueEntry(transactionID, false);
    }

    /**
     * Remove all AgentX request entries for the supplied transaction ID.
     *
     * @param transactionID
     *         a transaction ID.
     */
    public synchronized void removeAll(int transactionID) {
        getQueueEntry(transactionID, true);
    }

    private AgentXQueueEntry<?> getQueueEntry(int transactionID, boolean remove) {
        for (Iterator<AgentXQueueEntry<?>> it = queue.iterator(); it.hasNext(); ) {
            AgentXQueueEntry<?> entry = it.next();
            if (entry.transactionID == transactionID) {
                if (remove) {
                    it.remove();
                }
                return entry;
            }
        }
        return null;
    }

    class AgentXQueueEntry<A extends Address> implements Comparable<AgentXQueueEntry<A>> {

        private final int transactionID;
        private final LinkedList<AgentXPending<A>> requests;
        private int minTimeout = AgentXProtocol.MAX_TIMEOUT_SECONDS;
        private long timestamp = 0;

        AgentXQueueEntry(int transactionID) {
            this.transactionID = transactionID;
            this.requests = new LinkedList<>();
        }

        public synchronized final void addEntry(AgentXPending<A> pendingRequest) {
            this.requests.add(pendingRequest);
            if (minTimeout > pendingRequest.getTimeout()) {
                minTimeout = pendingRequest.getTimeout();
            }
        }

        public final void updateTimestamp() {
            this.timestamp = System.currentTimeMillis();
        }

        public final long getTimestamp() {
            return timestamp;
        }

        public final int getMinTimeout() {
            return minTimeout;
        }

        public boolean equals(Object obj) {
            if (obj instanceof AgentXQueueEntry) {
                AgentXQueueEntry<?> other = (AgentXQueueEntry<?>) obj;
                return ((transactionID == other.transactionID));
            }
            return false;
        }

        public int hashCode() {
            return transactionID;
        }

        public String toString() {
            return "AgentXQueueEntry[transactionID=" + transactionID + ",requests=" + requests + "]";
        }

        public int compareTo(AgentXQueueEntry<A> other) {
            return transactionID - other.transactionID;
        }

        public final synchronized AgentXPending<A> get(int sessionID, boolean remove) {
            for (Iterator<AgentXPending<A>> it = requests.iterator(); it.hasNext(); ) {
                AgentXPending<A> p = it.next();
                if (p.getSession().getSessionID() == sessionID) {
                    if (remove) {
                        it.remove();
                        if (requests.isEmpty()) {
                            queue.remove(this);
                        }
                    }
                    return p;
                }
            }
            return null;
        }

        public synchronized final boolean isEmpty() {
            return requests.isEmpty();
        }

        public synchronized final Collection<AgentXPending<A>> getPending() {
            LinkedList<AgentXPending<A>> pending = new LinkedList<>();
            for (AgentXPending<A> item : requests) {
                if (item.isPending()) {
                    pending.add(item);
                }
            }
            return pending;
        }
    }
}
