/*_############################################################################
  _## 
  _##  SNMP4J-AgentX - AgentXSubagent.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.subagent;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;

import org.snmp4j.PDU;
import org.snmp4j.TransportMapping;
import org.snmp4j.agent.*;
import org.snmp4j.agent.agentx.*;
import org.snmp4j.agent.agentx.event.PingEvent;
import org.snmp4j.agent.agentx.event.PingListener;
import org.snmp4j.agent.mo.*;
import org.snmp4j.agent.mo.lock.LockRequest;
import org.snmp4j.agent.mo.snmp.CoexistenceInfo;
import org.snmp4j.agent.request.*;
import org.snmp4j.log.LogAdapter;
import org.snmp4j.log.LogFactory;
import org.snmp4j.mp.SnmpConstants;
import org.snmp4j.smi.*;
import org.snmp4j.transport.*;
import org.snmp4j.agent.agentx.subagent.index.AnyNewIndexOID;
import org.snmp4j.agent.agentx.subagent.index.NewIndexOID;

import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.snmp4j.util.WorkerTask;

/**
 * The {@link AgentXSubagent} class implements the AgentX communication
 * for an AgentX subagent implementation and the agent configuration via {@link AgentConfigManager} although many
 * configurations that are necessary for a regular command responder and AgentX master agent are not applicable
 * for a subagent.
 * Since version 3.1.1 this class uses a {@link ConcurrentHashMap} to store AgentX sessions, which allows better
 * concurrent operations on agents using multiple sessions.
 *
 * @author Frank Fock
 * @version 3.2.0
 */
public class AgentXSubagent implements AgentXCommandListener, NotificationOriginator, TransportStateListener {

    private static final LogAdapter LOGGER = LogFactory.getLogger(AgentXSubagent.class);

    private final RequestFactory<AgentXCommandEvent<?>, AgentXResponsePDU, AgentXRequest> factory;
    private final AgentX agentX;
    /**
     * The request list.
     */
    protected final Map<RequestID, AgentXRequest> requestList;

    /**
     * The AgentX peer agents (masters).
     */
    protected final Map<Address, AgentXPeer<?>> peers = new LinkedHashMap<>(2);
    /**
     * The sessions hold by this sub-agent.
     */
    protected final Map<Integer, AgentXSession<?>> sessions = new ConcurrentHashMap<>(2);

    /**
     * Request handler for Get requests.
     */
    protected RequestHandler<AgentXRequest> requestHandlerGet;
    /**
     * Request handler for GetNext requests.
     */
    protected RequestHandler<AgentXRequest> requestHandlerGetNext;
    /**
     * Request handler for GetBulk requests.
     */
    protected RequestHandler<AgentXRequest> requestHandlerGetBulk;
    /**
     * Request handler for TestSet requests.
     */
    protected RequestHandler<AgentXRequest> requestHandlerTestSet;
    /**
     * Request handler for CommitSet requests.
     */
    protected RequestHandler<AgentXRequest> requestHandlerCommitSet;
    /**
     * Request handler for UndoSet requests.
     */
    protected RequestHandler<AgentXRequest> requestHandlerUndoSet;
    /**
     * Request handler for CleanupSet requests.
     */
    protected RequestHandler<AgentXRequest> requestHandlerCleanupSet;

    protected int nextTransactionID = 0;

    /**
     * Shared table support map with a key composed of {@code sessionID + "#" + context}.
     */
    protected Map<String, AgentXSharedMOTableSupport<?,?>> sharedMOTableSupport = new ConcurrentHashMap<>(5);

    private final OID subagentID;
    private final OctetString subagentDescr;

    private final long timeout = AgentXProtocol.DEFAULT_TIMEOUT_SECONDS * 1000;
    private byte defaultPriority = AgentXProtocol.DEFAULT_PRIORITY;

    private Timer pingTimer;
    private transient Vector<PingListener> pingListeners;
    private final SubagentXConfigManager configManager;

    /**
     * Creates a AgentX sub-agent using a {@link SubagentXConfigManager}.
     *
     * @param agentX
     *         the {@link AgentX} protocol to be used by this sub-agent.
     * @param subagentID
     *         the unique identifier for this sub-agent.
     * @param subagentDescr
     *         a textual description for this sub-agent.
     * @param configManager
     *         the configuration manager that organises how {@link ManagedObject}s of this sub-agent are initialized,
     *         persistently stored, and served to the agent.
     * @since 3.0
     */
    public AgentXSubagent(AgentX agentX,
                          OID subagentID, OctetString subagentDescr, SubagentXConfigManager configManager) {
        this.configManager = configManager;
        this.requestList = new ConcurrentHashMap<RequestID, AgentXRequest>(10);
        this.agentX = agentX;
        this.subagentID = subagentID;
        this.subagentDescr = subagentDescr;
        this.factory = new DefaultAgentXRequestFactory();
        requestHandlerGet = new GetRequestHandler();
        requestHandlerCleanupSet = new CleanupSetHandler();
        requestHandlerCommitSet = new CommitSetHandler();
        requestHandlerTestSet = new TestSetHandler();
        requestHandlerUndoSet = new UndoSetHandler();
        requestHandlerGetNext = new GetNextHandler();
        requestHandlerGetBulk = new GetBulkHandler();
        agentX.addCommandResponder(this);
    }

    /**
     * Sets the ping delay in seconds. If greater than zero, for each session
     * a ping PDU is sent to the master to validate the session regularly with
     * the specified delay. To monitor the ping requests, it is necessary to
     * add a {@link PingListener} with {@link #addPingListener}.
     *
     * @param seconds
     *         the delay. If zero or a negative value is supplied, no pings are sent
     */
    public void setPingDelay(int seconds) {
        if (pingTimer != null) {
            pingTimer.cancel();
            pingTimer = null;
        }
        if (seconds > 0) {
            pingTimer = new Timer();
            pingTimer.schedule(new PingTask(), seconds * 1000L, seconds * 1000L);
        }
    }

    @Override
    public <A extends Address> void processCommand(AgentXCommandEvent<A> event) {
        if (event.getCommand() != null) {
            event.setProcessed(true);
            Command command = new Command(event);
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Processing AgentX PDU: " + event.getCommand());
            }
            if (configManager.getWorkerPool() != null) {
                configManager.getWorkerPool().execute(command);
            } else {
                command.run();
            }
        }
    }

    /**
     * Get the next AgentX transaction ID.
     * @return
     *    a unique AgentX transaction ID for this sub-agent execution runtime.
     */
    protected synchronized int getNextTransactionID() {
        return nextTransactionID++;
    }

    /**
     * Close the AgentX session with the given session ID and reason.
     * @param sessionID
     *    an AgentX session ID. If such a session does not exist, {@link AgentXProtocol#AGENTX_NOT_OPEN} is returned.
     * @param reason
     *    the AgentX reason ({@link AgentXProtocol#REASON_SHUTDOWN}, {@link AgentXProtocol#REASON_TIMEOUTS},
     *    {@link AgentXProtocol#REASON_BY_MANAGER}, {@link AgentXProtocol#REASON_PROTOCOL_ERROR},
     *    {@link AgentXProtocol#REASON_PARSE_ERROR}, {@link AgentXProtocol#REASON_OTHER}) for closing the session.
     * @return
     *    the error status returned by the master agent or {@link AgentXProtocol#AGENTX_TIMEOUT} if the master did not
     *    respond in time.
     * @throws IOException
     *    if the IO operation failed.
     */
    protected int closeSession(int sessionID, byte reason) throws
            IOException {
        AgentXSession<?> session;
        session = removeSession(sessionID);
        if ((session == null) || (session.isClosed())) {
            return AgentXProtocol.AGENTX_NOT_OPEN;
        }
        session.setClosed(true);
        AgentXResponseEvent<?> resp = closeAgentXSession(session);
        if ((resp == null) || (resp.getResponse() == null)) {
            return AgentXProtocol.AGENTX_TIMEOUT;
        }
        return resp.getResponse().getErrorStatus();
    }

    private <A extends Address> AgentXResponseEvent<A> closeAgentXSession(AgentXSession<A> session) throws IOException {
        AgentXClosePDU closePDU = new AgentXClosePDU(AgentXProtocol.REASON_SHUTDOWN);
        closePDU.setSessionID(session.getSessionID());
        AgentXTarget<A> target = session.createAgentXTarget();
        return agentX.send(closePDU, target, session.getPeer().getTransport());
    }

    /**
     * Open a new AgentX session with the given transport and master address.
     * @param transport
     *    the {@link TransportMapping} to use.
     * @param masterAddress
     *    the AgentX master agent address to connect.
     * @param session
     *    a new {@link AgentXSession}. On success, the session ID will be set to the value returned by the master.
     * @param <A> type of {@link Address} for the session's master address.
     *
     * @return
     *    the error status returned by the master agent or {@link AgentXProtocol#AGENTX_TIMEOUT} if the master did not
     *    respond in time.
     * @throws IOException
     *    if the IO operation failed.
     */
    protected <A extends Address> int openSession(ConnectionOrientedTransportMapping<A> transport, A masterAddress,
                                                  AgentXSession<A> session) throws IOException {
        AgentXOpenPDU openPDU = new AgentXOpenPDU(0, getNextTransactionID(), 0, session.getTimeout(),
                subagentID, subagentDescr);
        // Make sure, we can send any packet to masterAddress:
        resumeConnecting(masterAddress, transport);
        AgentXResponseEvent<A> responseEvent = agentX.send(openPDU, session.createAgentXTarget(), transport);
        if (responseEvent.getResponse() == null) {
            LOGGER.error("Timeout on connection to master " + masterAddress);
        } else if (responseEvent.getResponse() != null) {
            AgentXResponsePDU response = responseEvent.getResponse();
            if (response.getErrorStatus() == AgentXProtocol.AGENTX_SUCCESS) {
                session.setSessionID(response.getSessionID());
            }
            return response.getErrorStatus();
        } else {
            LOGGER.error("Received packet on open PDU is not a response AgentX PDU: " + responseEvent);
        }
        return AgentXProtocol.AGENTX_TIMEOUT;
    }

    private static int getResponseStatus(AgentXResponseEvent<?> responseEvent) {
        if (responseEvent.getResponse() == null) {
            LOGGER.error("Timeout on connection to master " +
                    responseEvent.getTarget());
            return AgentXProtocol.AGENTX_TIMEOUT;
        } else if (responseEvent.getResponse() != null) {
            AgentXResponsePDU response = responseEvent.getResponse();
            return response.getErrorStatus();
        } else {
            LOGGER.error("Received packet on open PDU is not a response AgentX PDU: " +
                    responseEvent);
        }
        return AgentXProtocol.AGENTX_ERROR;
    }

    /**
     * Disconnect from master and suspend any sending of packets to the given master address, before actually
     * closing the transport used to communicate with the specified master.
     *
     * @param masterAddress
     *    the master's address to suspend and then disconnect.
     * @throws IOException
     *    if the closing of the connection fails.
     * @param <A>
     *    the {@link Address} type.
     * @since 3.2.0
     */
    @SuppressWarnings("unchecked")
    public <A extends Address> void disconnect(A masterAddress) throws IOException {
        AgentXPeer<?> peer = peers.remove(masterAddress);
        if (peer != null) {
            ConnectionOrientedTransportMapping<A> transport =
                    (ConnectionOrientedTransportMapping<A>) peer.getTransport();
            transport.suspendAddress(masterAddress);
            transport.close();
        }
    }

    /**
     * Resumes auto connection to the specified address using the given transport. This call does not actually
     * send any packets to the address, instead it simply (re)activates the mechanism to automatically establish
     * a connection to the specified address, when a new packet (i.e. openSessionPDU) is sent to the master address.
     * Thus, call {@link #openSession(ConnectionOrientedTransportMapping, Address, AgentXSession)} immediately
     * after calling this method to reconnect to a master, after a {@link #disconnect(Address)} with suspend or a
     * {@link #resetConnection(Address, boolean)} call.
     *
     * @param masterAddress
     *    the master's address to disconnect and optionally suspend.
     * @param transportMapping
     *    the {@link TransportMapping} to resume.
     * @param <A>
     *    the {@link Address} type.
     * @return
     *    {@code true} if the master address was suspended before and is resumed now, {@code false} if resuming the
     *    master address was not necessary (i.e. is already active).
     * @since 3.2.0
     */
    public <A extends Address> boolean resumeConnecting(A masterAddress,
                                                        ConnectionOrientedTransportMapping<A> transportMapping) {
        return transportMapping.resumeAddress(masterAddress);
    }

    /**
     * Reset a connection that is not working anymore either with trying to close all open session with a AgentXClosePdu
     * (when sendClosePdu is true) or silently without sending any messages. In both cases the connection associated
     * with the given master address is closed (with {@link #disconnect(Address)}) and all related sessions are
     * removed and closed too.
     * @param masterAddress
     *    the address of the peer to reset connection.
     * @param sendClosePdu
     *    {@code true} if this subagent should try to send AgentXClosePdus for each open session before resetting the
     *    connection of {@code false} if the connection should be closed silently. Close PDU will be sent only if the
     *    transport is still listening.
     * @param <A>
     *     the {@link Address} type of the connection.
     * @return
     *     {@code true} if the connection has been successfully reset, {@code false} otherwise.
     * @since 3.1.1
     */
    public <A extends Address> boolean resetConnection(A masterAddress, boolean sendClosePdu) {
        @SuppressWarnings("unchecked")
        AgentXPeer<A> peer = (AgentXPeer<A>)peers.get(masterAddress);
        TransportMapping<? super A> transport;
        if (peer != null) {
            List<AgentXSession<?>> sessionsToClose = new ArrayList<>();
            for (AgentXSession<?> session : sessions.values()) {
                if (peer.equals(session.getPeer())) {
                    sessionsToClose.add(session);
                }
            }
            transport = peer.getTransport();
            for (AgentXSession<?> sessionToClose : sessionsToClose) {
                AgentXSession<?> removedSession = removeSession(sessionToClose.getSessionID());
                sessionToClose.setClosed(true);
                if (removedSession != null) {
                    if (sendClosePdu && transport != null && transport.isListening()){
                        try {
                            closeAgentXSession(removedSession);
                        } catch (IOException e) {
                            LOGGER.error("IO exception while resetting AgentX session " + sessionToClose +
                                    ": " + e.getMessage(), e);
                        }
                    }
                    sessions.remove(sessionToClose.getSessionID());
                }
            }
            try {
                disconnect(masterAddress);
                return true;
            } catch (IOException e) {
                LOGGER.error("IO exception on closing connection of transport '"+transport+
                        "' to '"+masterAddress+"': "+e.getMessage(), e);
            }
        }
        return false;
    }

    /**
     * Connect to the given AgentX master using the specified local address.
     * @param masterAddress
     *    the (TCP) address of the master agent.
     * @param localAddress
     *    the local address. Use port 0 to let {@link AgentXSubagent} choose an available port.
     * @param session
     *    returns the {@link AgentXSession} information of the new AgentX session, i.e. the session ID returned by the
     *    master agent.
     * @param <A> address type to connect to.
     * @return
     *    {@link AgentXProtocol#AGENTX_SUCCESS} if the session has been established or an AgentX error status, if not.
     * @throws IOException
     *    if an IO operation failed.
     */
    public <A extends Address> int connect(A masterAddress, A localAddress, AgentXSession<A> session)
            throws IOException {
        return connect(masterAddress, localAddress, session, null);
    }

    /**
     * Connect to the given AgentX master using the specified local address.
     * @param masterAddress
     *    the (TCP) address of the master agent.
     * @param localAddress
     *    the local address. Use port 0 to let {@link AgentXSubagent} choose an available port.
     * @param session
     *    returns the {@link AgentXSession} information of the new AgentX session, i.e. the session ID returned by the
     *    master agent.
     * @param stateListeners
     *    an optional list of {@link TransportStateListener} to register for {@link TransportStateEvent}s before the
     *    connection is being established using a new {@link ConnectionOrientedTransportMapping}.
     * @param <A> address type to connect to.
     * @return
     *    {@link AgentXProtocol#AGENTX_SUCCESS} if the session has been established or an AgentX error status, if not.
     * @throws IOException
     *    if an IO operation failed.
     * @since 3.0.1
     */
    public <A extends Address> int connect(A masterAddress, A localAddress, AgentXSession<A> session,
                                           List<TransportStateListener> stateListeners)
            throws IOException {
        @SuppressWarnings("unchecked")
        AgentXPeer<A> peer = (AgentXPeer<A>)peers.get(masterAddress);
        ConnectionOrientedTransportMapping<A> transport;
        if (peer == null) {
            transport = addMaster(localAddress, stateListeners);
            peer = new AgentXPeer<>(transport, masterAddress);
        } else {
            transport = peer.getTransport();
            agentX.removeTransportMapping(transport);
            agentX.addTransportMapping(transport);
            if (!transport.isListening()) {
                transport.listen();
            }
        }
        peer.setTimeout(session.getTimeout());
        session.setPeer(peer);
        int status;
        try {
            status = openSession(transport, masterAddress, session);
            if (status != AgentXProtocol.AGENTX_TIMEOUT) {
                peers.put(masterAddress, peer);
                LOGGER.info("Added new peer address=" + masterAddress + ",peer=" + peer);
            }
        } catch (IOException ex) {
            LOGGER.error(ex);
            removeMaster(transport);
            return AgentXProtocol.AGENTX_ERROR;
        }
        if (status == AgentXProtocol.AGENTX_SUCCESS) {
            sessions.put(session.getSessionID(), session);
            LOGGER.info("Opened subagent session successfully: " + session);
        } else {
            removeMaster(transport);
        }
        return status;
    }

    /**
     * Close an AgentX session.
     * @param session
     *    a {@link AgentXSession}.
     * @param reason
     *    the AgentX reason ({@link AgentXProtocol#REASON_SHUTDOWN}, {@link AgentXProtocol#REASON_TIMEOUTS},
     *    {@link AgentXProtocol#REASON_BY_MANAGER}, {@link AgentXProtocol#REASON_PROTOCOL_ERROR},
     *    {@link AgentXProtocol#REASON_PARSE_ERROR}, {@link AgentXProtocol#REASON_OTHER}) for closing the session.
     * @return
     *    {@link AgentXProtocol#AGENTX_SUCCESS} if the session has been closed or an AgentX error status, if not.
     * @throws IOException
     *    if an IO operation failed.
     */
    public int close(AgentXSession<?> session, byte reason) throws IOException {
        return closeSession(session.getSessionID(), reason);
    }

    /**
     * Close all sessions hold by this sub-agent and return the AgentX status of the close operation together with the
     * {@link AgentXSession} object.
     * @param reason
     *    the AgentX reason ({@link AgentXProtocol#REASON_SHUTDOWN}, {@link AgentXProtocol#REASON_TIMEOUTS},
     *    {@link AgentXProtocol#REASON_BY_MANAGER}, {@link AgentXProtocol#REASON_PROTOCOL_ERROR},
     *    {@link AgentXProtocol#REASON_PARSE_ERROR}, {@link AgentXProtocol#REASON_OTHER}) for closing the session.
     * @return
     *    a map of the AgentXSession session objects hold by this agent and the corresponding AgentX error code. An
     *    error code of {@link AgentXProtocol#AGENTX_SUCCESS} indicates that this session was successfully closed.
     * @since 3.0.0
     */
    @Override
    public Map<AgentXSession<?>,Integer> closeAllSessions(byte reason) {
        HashMap<AgentXSession<?>, Integer> result = new HashMap<>(sessions.size());
        for (AgentXSession<?> agentXSession : sessions.values()) {
            int status;
            try {
                status = close(agentXSession, reason);
            } catch (IOException e) {
                LOGGER.error("IO exception while closing AgentX session "+agentXSession+
                        ": "+e.getMessage(), e);
                status = AgentXProtocol.AGENTX_ERROR;
            }
            result.put(agentXSession, status);
        }
        return result;
    }

    private AgentXSession<?> getSession(int sessionID) {
        return sessions.get(sessionID);
    }

    private AgentXSession<?> removeSession(int sessionID) {
        return sessions.remove(sessionID);
    }

    public void setDefaultPriority(byte priority) {
        this.defaultPriority = priority;
    }

    public byte getDefaultPriority() {
        return defaultPriority;
    }

    /**
     * Gets the priority with which the supplied managed object and
     * region should be registered at the master agent. Overwrite
     * this method to use individual priorities depending on the registered
     * region/managed object. The default implementation returns
     * {@link #getDefaultPriority()}.
     *
     * @param mo
     *         ManagedObject
     *         a managed object instance that manages {@code region}.
     * @param region
     *         the region to be registered.
     *
     * @return the priority between 0 and 255 (lower value results in higher priority).
     */
    protected byte getPriority(ManagedObject<?> mo, AgentXRegion region) {
        return defaultPriority;
    }

    /**
     * Registers the subagent regions at the master agent. It uses the
     * {@link AgentXSharedMOTableSupport} instances of {@link AgentXSharedMutableMOTable}
     * instances. For any other instances a support object instance will
     * be created for each session and context.
     *
     * @param session
     *         the session on whose behalf regions are registered.
     * @param context
     *         the context to use for registration.
     * @param sysUpTime
     *         if not {@code null}, the master agent's notion of the sysUpTime
     *         for the registered context is returned. The input value is always
     *         ignored!
     * @param registrationCallback
     *         a possibly {@code null} reference to a
     *         {@code RegistrationCallback} instance to handle registration
     *         events.
     */
    public void registerRegions(AgentXSession<?> session, OctetString context, TimeTicks sysUpTime,
                                RegistrationCallback registrationCallback) {
        MOServer server = getServer(context);
        if (server == null) {
            LOGGER.warn("No MOServer found for context '" + context + "'");
            return;
        }
        for (Iterator<Entry<MOScope, ManagedObject<?>>> it = server.iterator(); it.hasNext(); ) {
            Entry<MOScope, ManagedObject<?>> e = it.next();
            ManagedObject<? extends SubRequest<?>> mo = e.getValue();
            MOScope scope = e.getKey();
            if (context != null) {
                if ((scope instanceof MOContextScope) &&
                        (!context.equals(((MOContextScope) scope).getContext()))) {
                    continue;
                }
            }
            if (mo instanceof AgentXSharedMOTable) {
                registerSharedTableRows(session, context, registrationCallback, mo);
            } else {
                AgentXRegion region = new AgentXRegion(scope.getLowerBound(), scope.getUpperBound());
                if (mo instanceof MOScalar) {
                    region.setSingleOID(true);
                }
                region.setUpperIncluded(scope.isUpperIncluded());
                try {
                    int status = registerRegion(session, context, region,
                            getPriority(mo, region), sysUpTime);
                    if (status != AgentXProtocol.AGENTX_SUCCESS) {
                        if (LOGGER.isWarnEnabled()) {
                            LOGGER.warn("Failed to registered MO " + scope +
                                    " with status = " +
                                    status);
                        }
                    } else {
                        if (LOGGER.isInfoEnabled()) {
                            LOGGER.info("Registered MO " + scope + " successfully");
                        }
                    }
                    if (registrationCallback != null) {
                        registrationCallback.registrationEvent(context, mo, status);
                    }
                } catch (IOException ex) {
                    LOGGER.warn("Failed to register " + mo + " in context '" + context +
                            "' of session " + session);
                    if (registrationCallback != null) {
                        registrationCallback.registrationEvent(context, mo,
                                AgentXProtocol.AGENTX_ERROR);
                    }
                }
            }
        }
    }

    @SuppressWarnings("unchecked")
    private <A extends Address> void registerSharedTableRows(AgentXSession<A> session, OctetString context,
                                                             RegistrationCallback registrationCallback,
                                                             ManagedObject<? extends SubRequest<?>> mo) {
        AgentXSharedMOTableSupport<?,A> sharedTableSupport = null;
        if (mo instanceof AgentXSharedMutableMOTable) {
            sharedTableSupport = ((AgentXSharedMutableMOTable) mo).getAgentXSharedMOTableSupport();
        }
        String sharedTableSupportKey = session.getSessionID() + "#" + context;
        if (sharedTableSupport == null) {
            sharedTableSupport = (AgentXSharedMOTableSupport<?, A>) sharedMOTableSupport.get(sharedTableSupportKey);
        }
        if (sharedTableSupport == null) {
            sharedTableSupport = (AgentXSharedMOTableSupport<?, A>) createSharedTableSupport(session, context);
            sharedMOTableSupport.put(sharedTableSupportKey, sharedTableSupport);
        }
        registerSharedTableRows(session, context, (AgentXSharedMOTable)mo,
                registrationCallback, sharedTableSupport);
    }

    /**
     * Create a new {@link AgentXSharedMOTableSupport} instance for the given AgentX session and context.
     *
     * @param session
     *         an AgentXSession instance.
     * @param context
     *         a AgentX context.
     * @param <A> address type associated with the {@link AgentXSession} for which the shared table support is created.
     * @param <R> the {@link MOTableRow} type to support.
     *
     * @return a (new) AgentXSharedMOTableSupport instance.
     */
    protected <R extends MOTableRow, A extends Address> AgentXSharedMOTableSupport<R,?> createSharedTableSupport(
            AgentXSession<A> session, OctetString context ) {
        return new AgentXSharedMOTableSupport<R,A>(agentX, session, context);
    }

    /**
     * Registers the indexes and (row) regions of a shared table. This method
     * is called on behalf of {@link #registerRegions(org.snmp4j.agent.agentx.AgentXSession,
     * org.snmp4j.smi.OctetString, org.snmp4j.smi.TimeTicks, RegistrationCallback)}.
     *
     * @param session
     *         the session on whose behalf regions are registered.
     * @param context
     *         the context to use for registration.
     * @param mo
     *         the {@code AgentXSharedMOTable} instance to register.
     * @param registrationCallback
     *         if not {@code null} the callback is informed when registration
     *         of a row succeeded or failed.
     * @param <A> address type associated with the {@link AgentXSession} for which the shared table support is created.
     * @param <R> the {@link MOTableRow} type to support.
     * @param <M> the table model type managing the shared table rows.
     * @param <C> the column (base) type of the shared table.
     *
     * @deprecated Use {@link #registerSharedTableRows(org.snmp4j.agent.agentx.AgentXSession, org.snmp4j.smi.OctetString,
     * AgentXSharedMOTable, RegistrationCallback, AgentXSharedMOTableSupport)} instead. This version creates
     * a new table support object for each call (shared table) which is not recommended.
     */
    @Deprecated
    public <R extends MOTableRow, C extends MOColumn<?>, M extends MOTableModel<R>, A extends Address>
    void registerSharedTableRows(AgentXSession<A> session, OctetString context, final AgentXSharedMOTable<R, C, M> mo,
            RegistrationCallback registrationCallback) {
        registerSharedTableRows(session, context, mo, registrationCallback,
                new AgentXSharedMOTableSupport<R,A>(agentX, session, context));
    }

    /**
     * Registers the indexes and (row) regions of a shared table. This method
     * is called on behalf of {@link #registerRegions(org.snmp4j.agent.agentx.AgentXSession,
     * org.snmp4j.smi.OctetString, org.snmp4j.smi.TimeTicks, RegistrationCallback)}.
     *
     * @param session
     *         the session on whose behalf regions are registered.
     * @param context
     *         the context to use for registration.
     * @param mo
     *         the {@code AgentXSharedMOTable} instance to register.
     * @param registrationCallback
     *         if not {@code null} the callback is informed when registration
     *         of a row succeeded or failed.
     * @param sharedTableSupport
     *         the shared table support to be used for row registration. If {@code mo} has
     *         no table support instance and is a {@link AgentXSharedMutableMOTable} then its
     *         sharedTableSupport will be set to {@code sharedTableSupport}.
     * @param <R> the {@link MOTableRow} type to support.
     * @param <M> the table model type managing the shared table rows.
     * @param <C> the column (base) type of the shared table.
     *
     * @since 2.1
     */
    public <R extends MOTableRow, C extends MOColumn<?>, M extends MOTableModel<R>> void registerSharedTableRows(
            AgentXSession<?> session, OctetString context, final AgentXSharedMOTable<R,C,M> mo,
            RegistrationCallback registrationCallback, AgentXSharedMOTableSupport<R,?> sharedTableSupport) {
        synchronized (mo) {
            if ((mo instanceof AgentXSharedMutableMOTable) &&
                    (((AgentXSharedMutableMOTable) mo).getAgentXSharedMOTableSupport() == null)) {
                ((AgentXSharedMutableMOTable<R,C,M>)
                        mo).setAgentXSharedMOTableSupport(sharedTableSupport);
            }
            // decouple iterator from table modifications (may still fail if table
            // is being modified while row list is copied - if such a concurrency is
            // needed a table model must be used that returns an table independent
            // iterator.
            ArrayList<R> rows = new ArrayList<>(mo.getModel().getRowCount());
            for (Iterator<R> it = mo.getModel().iterator(); it.hasNext(); ) {
                rows.add(it.next());
            }
            for (R row : rows) {
                int retries = 0;
                int status;
                OID newIndex;
                do {
                    newIndex = (OID) row.getIndex().clone();
                    status = sharedTableSupport.allocateIndex(context, mo.getIndexDef(),
                            (byte) AgentXSharedMOTableSupport.INDEX_MODE_ALLOCATE,
                            newIndex);
                }
                while ((registrationCallback != null) &&
                        registrationCallback.tableRegistrationEvent(context,
                                mo, row, true, status, retries++));
                if (status == AgentXProtocol.AGENTX_SUCCESS) {
                    if ((newIndex instanceof AnyNewIndexOID) ||
                            (newIndex instanceof NewIndexOID)) {
                        if (mo instanceof AgentXSharedMutableMOTable) {
                            ((AgentXSharedMutableMOTable) mo).
                                    changeRowIndex(row.getIndex(), newIndex);
                        }
                        break;
                    }
                    status = sharedTableSupport.registerRow(mo, row);
                    if (status != AgentXProtocol.AGENTX_SUCCESS) {
                        sharedTableSupport.deallocateIndex(context, mo.getIndexDef(),
                                row.getIndex());
                        LOGGER.warn("Failed to register row with " + status + " for " +
                                row);
                    }
                    if (registrationCallback != null) {
                        registrationCallback.tableRegistrationEvent(context,
                                mo, row, false, status, retries);
                    }
                } else {
                    LOGGER.warn("Failed to allocate index with " + status + " for row " +
                            row);
                }
            }
        }
    }

    protected <A extends Address> int registerRegion(AgentXSession<A> session,
                                 OctetString context, AgentXRegion region,
                                 byte priority,
                                 TimeTicks sysUpTime) throws IOException {
        if ((session == null) || (session.isClosed())) {
            return AgentXProtocol.AGENTX_NOT_OPEN;
        }
        long t = (this.timeout == 0) ? session.getTimeout() * 1000L : this.timeout;
        AgentXRegisterPDU pdu =
                new AgentXRegisterPDU(context, region.getLowerBound(), priority,
                        region.getRangeSubID(),
                        region.getUpperBoundSubID());
        pdu.setSessionAttributes(session);
        AgentXResponseEvent<A> event = agentX.send(pdu, new AgentXTarget<>(session.getPeer().getAddress(), t),
                        session.getPeer().getTransport());
        if ((sysUpTime != null) && (event.getResponse() != null)) {
            sysUpTime.setValue(event.getResponse().getSysUpTime() & 0xFFFFFFFFL);
        }
        return getResponseStatus(event);
    }

    protected <A extends Address> int unregisterRegion(AgentXSession<A> session, OctetString context,
                                                       AgentXRegion region, byte timeout) throws IOException {
        if ((session == null) || (session.isClosed())) {
            return AgentXProtocol.AGENTX_NOT_OPEN;
        }
        byte t = (timeout == 0) ? session.getTimeout() : timeout;
        AgentXUnregisterPDU pdu =
                new AgentXUnregisterPDU(context, region.getLowerBound(), t,
                        region.getRangeSubID(),
                        region.getUpperBoundSubID());
        pdu.setSessionAttributes(session);
        AgentXResponseEvent<A> event = agentX.send(pdu, new AgentXTarget<>(session.getPeer().getAddress(),
                        this.timeout), session.getPeer().getTransport());
        return getResponseStatus(event);
    }

    protected <A extends Address> ConnectionOrientedTransportMapping<A> addMaster(A localAddress,
                                                                List<TransportStateListener> stateListeners)
            throws IOException {
        ConnectionOrientedTransportMapping<A> transport = (ConnectionOrientedTransportMapping<A>)
                 TransportMappings.getInstance().createTransportMapping(localAddress);
        transport.setServerEnabled(false);
        transport.setConnectionTimeout(0);
        transport.setMessageLengthDecoder(new AgentXProtocol());
        if (stateListeners != null) {
            for (TransportStateListener transportStateListener : stateListeners) {
                transport.addTransportStateListener(transportStateListener);
            }
        }
        agentX.addTransportMapping(transport);
        transport.listen();
        return transport;
    }

    protected void removeMaster(TransportMapping<?> transport) {
        agentX.removeTransportMapping(transport);
        try {
            transport.close();
        } catch (IOException ex) {
            LOGGER.warn("Closing transport mapping " + transport + " failed with: " + ex.getMessage());
        }
    }

    public synchronized MOServer getServer(OctetString context) {
        MOServer s = configManager.getServer(context);
        if ((s != null) && s.isContextSupported(context)) {
            return s;
        }
        return null;
    }

    public synchronized Collection<OctetString> getContexts() {
        LinkedList<OctetString> allContexts = new LinkedList<OctetString>();
        for (MOServer s : configManager.getServers()) {
            OctetString[] contexts = s.getContexts();
            allContexts.addAll(Arrays.asList(contexts));
        }
        return allContexts;
    }

    public void dispatchCommand(AgentXCommandEvent<?> cmd) {
        boolean pendingSessionClose = false;
        if (cmd.getCommand().isConfirmedPDU()) {
            AgentXRequest request = null;
            MOServer server = null;
            int type = cmd.getCommand().getType();
            switch (type) {
                case AgentXPDU.AGENTX_GET_PDU: {
                    request = factory.createRequest(cmd, null);
                    server = getServer(request.getContext());
                    requestHandlerGet.processPdu(request, server);
                    break;
                }
                case AgentXPDU.AGENTX_GETNEXT_PDU: {
                    request = factory.createRequest(cmd, null);
                    server = getServer(request.getContext());
                    requestHandlerGetNext.processPdu(request, server);
                    break;
                }
                case AgentXPDU.AGENTX_GETBULK_PDU: {
                    request = factory.createRequest(cmd, null);
                    server = getServer(request.getContext());
                    requestHandlerGetBulk.processPdu(request, server);
                    break;
                }
                case AgentXPDU.AGENTX_TESTSET_PDU: {
                    request = factory.createRequest(cmd, null);
                    request.setPhase(Request.PHASE_2PC_PREPARE);
                    server = getServer(request.getContext());
                    requestHandlerTestSet.processPdu(request, server);
                    requestList.put(createRequestID(cmd), request);
                    break;
                }
                case AgentXPDU.AGENTX_COMMITSET_PDU:
                case AgentXPDU.AGENTX_UNDOSET_PDU:
                case AgentXPDU.AGENTX_CLEANUPSET_PDU: {
                    RequestID reqID = createRequestID(cmd);
                    request = requestList.get(reqID);
                    if (request == null) {
                        LOGGER.error("Request with ID " + reqID + " not found in request list");
                        request = new AgentXRequest(cmd);
                        request.setErrorStatus(AgentXProtocol.AGENTX_PROCESSING_ERROR);
                        break;
                    }
                    server = getServer(request.getContext());
                    switch (type) {
                        case AgentXPDU.AGENTX_COMMITSET_PDU:
                            request.setPhase(Request.PHASE_2PC_COMMIT);
                            requestHandlerCommitSet.processPdu(request, server);
                            break;
                        case AgentXPDU.AGENTX_UNDOSET_PDU:
                            request.setPhase(Request.PHASE_2PC_UNDO);
                            requestHandlerUndoSet.processPdu(request, server);
                            break;
                        case AgentXPDU.AGENTX_CLEANUPSET_PDU:
                            request.setPhase(Request.PHASE_2PC_CLEANUP);
                            requestHandlerCleanupSet.processPdu(request, server);
                            break;
                        default: {
                            LOGGER.fatal("Internal error");
                        }
                    }
                    if (cmd.getCommand().getType() != AgentXPDU.AGENTX_COMMITSET_PDU) {
                        // remove request from request list
                        requestList.remove(reqID);
                    }
                    break;
                }
                case AgentXPDU.AGENTX_CLOSE_PDU: {
                    AgentXSession<?> session = removeSession(cmd.getCommand().getSessionID());
                    if (session != null) {
                        session.setClosed(true);
                        pendingSessionClose = true;
                    }
                    break;
                }
                default: {
                    LOGGER.error("Unhandled PDU type: " + cmd.getCommand());
                    request = new AgentXRequest(cmd);
                    request.setErrorStatus(AgentXProtocol.AGENTX_PROCESSING_ERROR);
                }
            }
            if (request != null) {
                // Since this is an AgentX subagent it only processes a single phase at
                // once.
                if ((type != AgentXPDU.AGENTX_CLEANUPSET_PDU) &&
                        request.isPhaseComplete()) {
                    // send response
                    sendResponse(cmd, request);
                }
                if (server != null) {
                    release(server, request);
                }
            }
            if (pendingSessionClose) {
                try {
                    disconnect((TcpAddress)cmd.getPeerAddress());
                } catch (IOException ex) {
                    LOGGER.error("Failed to disconnect from master at " +
                            cmd.getPeerAddress() + ": " + ex.getMessage(), ex);
                }
            }
        } else {
            processResponse(cmd);
        }
    }

    protected <A extends Address> void sendResponse(AgentXCommandEvent<A> cmd, AgentXRequest request) {
        AgentXMessageDispatcher dispatcher = cmd.getDispatcher();
        AgentXResponsePDU response = request.getResponsePDU();
        if (response != null) {
            AgentXPDU rpdu = cmd.getCommand();
            response.setSessionID(rpdu.getSessionID());
            response.setTransactionID(rpdu.getTransactionID());
            response.setByteOrder(rpdu.getByteOrder());
            response.setPacketID(rpdu.getPacketID());
            // only send a response if required
            try {
                dispatcher.send(cmd.getPeerTransport(), cmd.getPeerAddress(), response, null);
            } catch (IOException ex) {
                LOGGER.warn("Failed to send AgentX response to '" +
                        cmd.getPeerAddress() + "' with error: " + ex.getMessage());
            }
        }
    }

    protected void release(MOServer server, AgentXRequest req) {
        for (Iterator<AgentXRequest.AgentXSubRequest> it = req.iterator(); it.hasNext(); ) {
            SubRequest<?> sreq = it.next();
            if (sreq.getTargetMO() != null) {
                server.unlockNow(req, sreq.getTargetMO());
            }
        }
    }

    private static RequestID createRequestID(AgentXCommandEvent<?> cmd) {
        return new RequestID(cmd.getPeerAddress(),
                cmd.getCommand().getSessionID(),
                cmd.getCommand().getTransactionID());

    }

    protected void processResponse(AgentXCommandEvent<?> cmd) {
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Received response " + cmd);
        }
    }

    protected void processNextSubRequest(AgentXRequest request, MOServer server, OctetString context,
                                         AgentXRequest.AgentXSubRequest sreq)
            throws NoSuchElementException {
        // We can be sure to have a default context scope here because
        // the inner class AgentXSubRequest creates it!
        DefaultMOContextScope scope =
                (DefaultMOContextScope) sreq.getScope();
        MOQuery query = sreq.getQuery();
        if (query == null) {
            query = new MOQueryWithSource(scope, false, request);
        }
        LockRequest lockRequest =
                new LockRequest(request, request.getSource().createAgentXPeer().getTimeoutMillis());
        while (!sreq.getStatus().isProcessed()) {
            MOServerLookupEvent lookupEvent = new MOServerLookupEvent(this, null,
                    query, MOServerLookupEvent.IntendedUse.getNext, true);
            sreq.setLookupEvent(lookupEvent);
            GenericManagedObject mo = server.lookup(query, lockRequest, lookupEvent, GenericManagedObject.class);
            if (mo == null) {
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("EndOfMibView at scope=" + query.getScope() + " and query " + query);
                }
                sreq.getVariableBinding().setVariable(Null.endOfMibView);
                sreq.getStatus().setPhaseComplete(true);
                break;
            }
            try {
                if (!mo.next(sreq, null)) {
                    // We can be sure to have a default context scope here because
                    // the inner class SnmpSubRequest creates it!
                    // don't forget to update query:
                    sreq.getVariableBinding().setVariable(Null.instance);
                    scope.subtractScope(mo.getScope());
                    // query is updated automatically because scope is updated.
                    query.subtractScope(mo.getScope());
                }
            } catch (Exception moex) {
                if (LOGGER.isDebugEnabled()) {
                    moex.printStackTrace();
                }
                LOGGER.warn(moex);
                if (sreq.getStatus().getErrorStatus() == PDU.noError) {
                    sreq.getStatus().setErrorStatus(PDU.genErr);
                }
            }
            finally {
                lookupEvent.completedUse(sreq);
                unlockManagedObjectIfLockedByLookup(server, mo, lockRequest);
            }
        }
    }

    /**
     * Sends notifications (traps) to all appropriate notification targets
     * through the master agent.
     *
     * @param context
     *         the context name of the context on whose behalf this
     *         notification has been generated.
     * @param notificationID
     *         the object ID that uniquely identifies this
     *         notification. For SNMPv1 traps, the notification ID has to be build
     *         using the rules provided by RFC 2576.
     * @param vbs
     *         an array of {@code VariableBinding} instances
     *         representing the payload of the notification.
     *
     * @return an {@link AgentXResponseEvent} instance or {@code null} if the
     * notification request timed out.
     */
    public Object notify(OctetString context, OID notificationID, VariableBinding[] vbs) {
        return notify(context, notificationID, null, vbs);
    }

    public Object notify(OctetString context, OID notificationID, TimeTicks sysUpTime, VariableBinding[] vbs) {
        AgentXSession<?> session = firstSession();
        AgentXResponseEvent<?> agentXResponse;
        try {
            agentXResponse = notify(session, context, notificationID, sysUpTime, vbs);
            if ((agentXResponse == null) || (agentXResponse.getResponse() == null)) {
                LOGGER.warn("Timeout on sending notification in context '" + context +
                        "' with ID '" + notificationID + "' and payload " +
                        Arrays.asList(vbs));
                return null;
            }
            return agentXResponse;
        } catch (IOException ex) {
            LOGGER.error("Failed to send notification in context '" + context +
                    "' with ID '" + notificationID + "' and payload " +
                    Arrays.asList(vbs) + ", reason is: " + ex.getMessage());
            return null;
        }
    }

    /**
     * Returns the first session that have been opened by this subagent and is
     * still open. If no open session exists, {@code null} is returned.
     *
     * @return an {@code AgentXSession}.
     */
    public final AgentXSession<?> firstSession() {
        if (!sessions.isEmpty()) {
            return sessions.values().iterator().next();
        }
        return null;
    }

    /**
     * Send an AgentX notification to the master which will then be forwarded to trap/notification targets of the
     * master.
     *
     * @param session
     *    the {@link AgentXSession} to be used.
     * @param context
     *    the SNMPv3 (AgentX) context of the notification.
     * @param notificationID
     *    the {@link OID} the identifies the notification as defined in the corresponding MIB.
     * @param sysUpTime
     *    the current notion of the sub-agents up-time.
     * @param vbs
     *    a notification payload as an array of {@link VariableBinding} instances.
     *    {@link AgentXProtocol#AGENTX_SUCCESS} if the session has been closed or an AgentX error status, if not.
     * @param <A> address type associated with the {@link AgentXSession} for which the notification is created.
     * @return a {@link AgentXResponseEvent} object that provides detailed information about the response returned
     *    by the AgentX peer.
     * @throws IOException
     *    if an IO operation failed.
     */
    public <A extends Address> AgentXResponseEvent<A> notify(AgentXSession<A> session, OctetString context,
                                                             OID notificationID, TimeTicks sysUpTime,
                                                             VariableBinding[] vbs) throws IOException {
        int offset = 1;
        if (sysUpTime != null) {
            offset = 2;
        }
        VariableBinding[] notifyVBs = new VariableBinding[vbs.length + offset];
        if (sysUpTime != null) {
            notifyVBs[0] = new VariableBinding(SnmpConstants.sysUpTime, sysUpTime);
        }
        notifyVBs[offset - 1] =
                new VariableBinding(SnmpConstants.snmpTrapOID, notificationID);
        System.arraycopy(vbs, 0, notifyVBs, offset, vbs.length);
        AgentXNotifyPDU notifyPDU = new AgentXNotifyPDU(context, notifyVBs);
        notifyPDU.setSessionAttributes(session);
        notifyPDU.setTransactionID(getNextTransactionID());
        return agentX.send(notifyPDU, session.createAgentXTarget(), session.getPeer().getTransport());
    }

    public <A extends Address> int addAgentCaps(AgentXSession<A> session, OctetString context, OID id,
                                                OctetString descr) {
        AgentXAddAgentCapsPDU pdu = new AgentXAddAgentCapsPDU(context, id, descr);
        pdu.setSessionAttributes(session);
        try {
            AgentXResponseEvent<A> resp =
                    agentX.send(pdu, session.createAgentXTarget(), session.getPeer().getTransport());
            if (resp.getResponse() == null) {
                return AgentXProtocol.AGENTX_TIMEOUT;
            }
            return resp.getResponse().getErrorStatus();
        } catch (IOException ex) {
            LOGGER.error("Failed to send AgentX AddAgentCaps PDU " + pdu +
                    " because: " + ex.getMessage(), ex);
            return AgentXProtocol.AGENTX_NOT_OPEN;
        }
    }

    public <A extends Address> int removeAgentCaps(AgentXSession<A> session, OctetString context, OID id) {
        AgentXRemoveAgentCapsPDU pdu = new AgentXRemoveAgentCapsPDU(context, id);
        pdu.setSessionAttributes(session);
        try {
            AgentXResponseEvent<A> resp = agentX.send(pdu, session.createAgentXTarget(),
                    session.getPeer().getTransport());
            return resp.getResponse().getErrorStatus();
        } catch (IOException ex) {
            LOGGER.error("Failed to send AgentX RemoveAgentCaps PDU " + pdu +
                    " because: " + ex.getMessage(), ex);
            return AgentXProtocol.AGENTX_NOT_OPEN;
        }
    }

    public void addPingListener(PingListener l) {
        if (pingListeners == null) {
            pingListeners = new Vector<PingListener>();
        }
        pingListeners.add(l);
    }

    public void removePingListener(PingListener l) {
        if (pingListeners != null) {
            synchronized (pingListeners) {
                pingListeners.remove(l);
            }
        }
    }

    protected void firePinged(PingEvent<?> event) {
        final Vector<PingListener> listeners = pingListeners;
        if (listeners != null) {
            synchronized (listeners) {
                for (PingListener listener : listeners) {
                    listener.pinged(event);
                }
            }
        }
    }

    private static void initRequestPhase(Request<?,?,?> request) {
        if (request.getPhase() == Request.PHASE_INIT) {
            request.nextPhase();
        }
    }

    @Override
    public void connectionStateChanged(TransportStateEvent transportStateEvent) {

    }

    static class GetRequestHandler implements RequestHandler<AgentXRequest> {

        public boolean isSupported(int pduType) {
            return pduType == AgentXPDU.AGENTX_GET_PDU;
        }

        public void processPdu(AgentXRequest request, MOServer server) {
            initRequestPhase(request);
            LockRequest lockRequest =
                    new LockRequest(request, request.getSource().createAgentXPeer().getTimeoutMillis());
            try {
                Iterator<AgentXRequest.AgentXSubRequest> it = request.iterator();
                while (it.hasNext()) {
                    AgentXRequest.AgentXSubRequest sreq = it.next();
                    DefaultMOQuery query = new MOQueryWithSource((MOContextScope) sreq.getScope(),
                            false, request);
                    MOServerLookupEvent lookupEvent = new MOServerLookupEvent(this, null,
                            query, MOServerLookupEvent.IntendedUse.get, true);
                    sreq.setLookupEvent(lookupEvent);
                    ManagedObject<SubRequest<?>> mo =
                            server.lookup(query, lockRequest, lookupEvent, GenericManagedObject.class);
                    if (mo == null) {
                        sreq.getVariableBinding().setVariable(Null.noSuchObject);
                        sreq.getStatus().setPhaseComplete(true);
                    }
                    else {
                        sreq.setTargetMO(mo);
                    }
                }
                it = request.iterator();
                while ((!request.isPhaseComplete()) && (it.hasNext())) {
                    AgentXRequest.AgentXSubRequest sreq = it.next();
                    if (sreq.isComplete()) {
                        continue;
                    }
                    ManagedObject<? super AgentXRequest.AgentXSubRequest> mo = sreq.getTargetMO();
                    MOServerLookupEvent lookupEvent = sreq.getLookupEvent();
                    try {
                        mo.get(sreq);
                    } catch (Exception moex) {
                        if (LOGGER.isDebugEnabled()) {
                            moex.printStackTrace();
                        }
                        LOGGER.warn(moex);
                        if (sreq.getStatus().getErrorStatus() == PDU.noError) {
                            sreq.getStatus().setErrorStatus(PDU.genErr);
                        }
                    }
                    finally {
                        if (lookupEvent != null) {
                            lookupEvent.completedUse(sreq);
                        }
                        unlockManagedObjectIfLockedByLookup(server, mo, lockRequest);
                    }
                }
            } catch (NoSuchElementException nsex) {
                if (LOGGER.isDebugEnabled()) {
                    nsex.printStackTrace();
                }
                LOGGER.error("SubRequest not found");
                request.setErrorStatus(PDU.genErr);
            }
        }
    }

    class GetNextHandler implements RequestHandler<AgentXRequest> {

        public void processPdu(AgentXRequest request, MOServer server) {
            initRequestPhase(request);
            OctetString context = request.getContext();
            try {
                Iterator<AgentXRequest.AgentXSubRequest> it = request.iterator();
                while (it.hasNext()) {
                    AgentXRequest.AgentXSubRequest sreq = it.next();
                    processNextSubRequest(request, server, context, sreq);
                }
            } catch (NoSuchElementException nsex) {
                if (LOGGER.isDebugEnabled()) {
                    nsex.printStackTrace();
                }
                LOGGER.error("SubRequest not found");
                request.setErrorStatus(PDU.genErr);
            }
        }


        public boolean isSupported(int pduType) {
            return (pduType == PDU.GETNEXT);
        }

    }

    class GetBulkHandler implements RequestHandler<AgentXRequest> {

        public void processPdu(AgentXRequest request, MOServer server) {
            initRequestPhase(request);
            OctetString context = request.getContext();
            int nonRep = request.getNonRepeaters();
            try {
                Iterator<AgentXRequest.AgentXSubRequest> it = request.iterator();
                int i = 0;
                // non repeaters
                for (; ((i < nonRep) && it.hasNext()); i++) {
                    AgentXRequest.AgentXSubRequest sreq = it.next();
                    processNextSubRequest(request, server, context, sreq);
                }
                // repetitions
                for (; it.hasNext(); i++) {
                    AgentXRequest.AgentXSubRequest sreq = it.next();
                    processNextSubRequest(request, server, context, sreq);
                }
            } catch (NoSuchElementException nsex) {
                if (LOGGER.isDebugEnabled()) {
                    nsex.printStackTrace();
                }
                LOGGER.error("SubRequest not found");
                request.setErrorStatus(PDU.genErr);
            }

        }

        public boolean isSupported(int pduType) {
            return (pduType == PDU.GETBULK);
        }

    }


    static class TestSetHandler implements RequestHandler<AgentXRequest> {

        public void processPdu(AgentXRequest request, MOServer server) {
            try {
                Iterator<AgentXRequest.AgentXSubRequest> it = request.iterator();
                LockRequest lockRequest =
                        new LockRequest(request, request.getSource().createAgentXPeer().getTimeoutMillis());
                while ((!request.isPhaseComplete()) && (it.hasNext())) {
                    AgentXRequest.AgentXSubRequest sreq = it.next();
                    if (sreq.isComplete()) {
                        continue;
                    }
                    DefaultMOQuery query =
                            new MOQueryWithSource((MOContextScope) sreq.getScope(), true, request);
                    MOServerLookupEvent lookupEvent = new MOServerLookupEvent(this, null,
                            query, MOServerLookupEvent.IntendedUse.prepare, true);
                    sreq.setLookupEvent(lookupEvent);
                    ManagedObject<? super AgentXRequest.AgentXSubRequest> mo =
                            server.lookup(query, lockRequest, lookupEvent, GenericManagedObject.class);
                    if (mo == null) {
                        sreq.getStatus().setErrorStatus(PDU.notWritable);
                        break;
                    }
                    sreq.setTargetMO(mo);
                    if (lockRequest.getLockRequestStatus() == LockRequest.LockStatus.lockTimedOut) {
                        LOGGER.warn("TestSet request " + request +
                                " failed because " + mo + " could not be locked");
                        if (sreq.getStatus().getErrorStatus() == PDU.noError) {
                            sreq.getStatus().setErrorStatus(PDU.genErr);
                        }
                    }
                }
                it = request.iterator();
                while ((!request.isPhaseComplete()) && (it.hasNext())) {
                    AgentXRequest.AgentXSubRequest sreq = it.next();
                    if (sreq.isComplete()) {
                        continue;
                    }
                    ManagedObject<? super AgentXRequest.AgentXSubRequest> mo = sreq.getTargetMO();
                    MOServerLookupEvent lookupEvent = sreq.getLookupEvent();
                    try {
                        mo.prepare(sreq);
                        sreq.getStatus().setPhaseComplete(true);
                        lookupEvent.completedUse(sreq);
                    } catch (Exception moex) {
                        if (sreq.getStatus().getErrorStatus() == PDU.noError) {
                            sreq.getStatus().setErrorStatus(PDU.genErr);
                        }
                        LOGGER.error("Exception occurred while preparing SET request, " +
                                "returning genErr: " + moex.getMessage(), moex);
                    }
                    finally {
                        lookupEvent.completedUse(sreq);
                    }
                }
            } catch (NoSuchElementException nsex) {
                if (LOGGER.isDebugEnabled()) {
                    nsex.printStackTrace();
                }
                LOGGER.error("Cannot find sub-request: ", nsex);
                request.setErrorStatus(PDU.genErr);
            }
        }

        public boolean isSupported(int pduType) {
            return (pduType == AgentXPDU.AGENTX_TESTSET_PDU);
        }
    }

    class UndoSetHandler implements RequestHandler<AgentXRequest> {

        public void processPdu(AgentXRequest request, MOServer server) {
            try {
                Iterator<AgentXRequest.AgentXSubRequest> it = request.iterator();
                while (it.hasNext()) {
                    AgentXRequest.AgentXSubRequest sreq = it.next();
                    if (sreq.isComplete()) {
                        continue;
                    }
                    ManagedObject<? super AgentXRequest.AgentXSubRequest> mo =
                            lookupManagedObjectFromCachedTargetMO(server, sreq, true,
                                    MOServerLookupEvent.IntendedUse.undo);
                    if (mo == null) {
                        sreq.getStatus().setErrorStatus(PDU.undoFailed);
                    }
                }
                it = request.iterator();
                while ((!request.isPhaseComplete()) && (it.hasNext())) {
                    AgentXRequest.AgentXSubRequest sreq = it.next();
                    if (sreq.isComplete()) {
                        continue;
                    }
                    ManagedObject<? super AgentXRequest.AgentXSubRequest> mo = sreq.getTargetMO();
                    MOServerLookupEvent lookupEvent = sreq.getLookupEvent();
                    try {
                        mo.undo(sreq);
                        sreq.getStatus().setPhaseComplete(true);
                    } catch (Exception moex) {
                        if (LOGGER.isDebugEnabled()) {
                            moex.printStackTrace();
                        }
                        LOGGER.error(moex);
                        if (sreq.getStatus().getErrorStatus() == PDU.noError) {
                            sreq.getStatus().setErrorStatus(PDU.undoFailed);
                        }
                    }
                    finally {
                        if (lookupEvent != null &&
                                lookupEvent.getIntendedUse() == MOServerLookupEvent.IntendedUse.undo) {
                            lookupEvent.completedUse(sreq);
                        }
                    }
                }
            } catch (NoSuchElementException nsex) {
                if (LOGGER.isDebugEnabled()) {
                    nsex.printStackTrace();
                }
                LOGGER.error("Cannot find sub-request: ", nsex);
                request.setErrorStatus(PDU.genErr);
            }
        }

        public boolean isSupported(int pduType) {
            return (pduType == AgentXPDU.AGENTX_UNDOSET_PDU);
        }
    }

    private ManagedObject<? super AgentXRequest.AgentXSubRequest> lookupManagedObjectFromCachedTargetMO(
            MOServer server, AgentXRequest.AgentXSubRequest sreq, boolean isWriteAccessIntended,
            MOServerLookupEvent.IntendedUse intendedUse) {
        ManagedObject<? super AgentXRequest.AgentXSubRequest> mo = sreq.getTargetMO();
        if (mo == null) {
            DefaultMOQuery query =
                    new MOQueryWithSource((MOContextScope) sreq.getScope(), isWriteAccessIntended, sreq);
            MOServerLookupEvent lookupEvent = new MOServerLookupEvent(this, null,
                    query, intendedUse, true);
            sreq.setLookupEvent(lookupEvent);
            mo = server.lookup(query, null, lookupEvent, GenericManagedObject.class);
        }
        return mo;
    }

    class CommitSetHandler implements RequestHandler<AgentXRequest> {

        public void processPdu(AgentXRequest request, MOServer server) {
            try {
                Iterator<AgentXRequest.AgentXSubRequest> it = request.iterator();
                while ((!request.isPhaseComplete()) && (it.hasNext())) {
                    AgentXRequest.AgentXSubRequest sreq = it.next();
                    if (sreq.isComplete()) {
                        continue;
                    }
                    ManagedObject<? super AgentXRequest.AgentXSubRequest> mo =
                            lookupManagedObjectFromCachedTargetMO(server, sreq, true,
                                    MOServerLookupEvent.IntendedUse.commit);
                    if (mo == null) {
                        sreq.getStatus().setErrorStatus(PDU.commitFailed);
                    }
                }
                it = request.iterator();
                while ((!request.isPhaseComplete()) && (it.hasNext())) {
                    AgentXRequest.AgentXSubRequest sreq = it.next();
                    if (sreq.isComplete()) {
                        continue;
                    }
                    ManagedObject<? super AgentXRequest.AgentXSubRequest> mo = sreq.getTargetMO();
                    MOServerLookupEvent lookupEvent = sreq.getLookupEvent();
                    try {
                        mo.commit(sreq);
                        sreq.getStatus().setPhaseComplete(true);
                    } catch (Exception moex) {
                        if (LOGGER.isDebugEnabled()) {
                            moex.printStackTrace();
                        }
                        LOGGER.error(moex);
                        if (sreq.getStatus().getErrorStatus() == PDU.noError) {
                            sreq.getStatus().setErrorStatus(PDU.commitFailed);
                        }
                    }
                    finally {
                        if (lookupEvent != null &&
                                lookupEvent.getIntendedUse() == MOServerLookupEvent.IntendedUse.commit) {
                            lookupEvent.completedUse(sreq);
                        }
                    }
                }
            } catch (NoSuchElementException nsex) {
                if (LOGGER.isDebugEnabled()) {
                    nsex.printStackTrace();
                }
                LOGGER.error("Cannot find sub-request: ", nsex);
                request.setErrorStatus(PDU.genErr);
            }
        }

        public boolean isSupported(int pduType) {
            return (pduType == AgentXPDU.AGENTX_COMMITSET_PDU);
        }

    }

    class CleanupSetHandler implements RequestHandler<AgentXRequest> {

        public void processPdu(AgentXRequest request, MOServer server) {
            try {
                Iterator<AgentXRequest.AgentXSubRequest> it = request.iterator();
                while (it.hasNext()) {
                    AgentXRequest.AgentXSubRequest sreq = it.next();
                    if (sreq.isComplete()) {
                        continue;
                    }
                    ManagedObject<? super AgentXRequest.AgentXSubRequest> mo =
                            lookupManagedObjectFromCachedTargetMO(server, sreq, false,
                                    MOServerLookupEvent.IntendedUse.cleanUp);
                    if (mo == null) {
                        sreq.completed();
                        continue;
                    }
                    MOServerLookupEvent lookupEvent = sreq.getLookupEvent();
                    try {
                        mo.cleanup(sreq);
                        sreq.getStatus().setPhaseComplete(true);
                    } catch (Exception moex) {
                        if (LOGGER.isDebugEnabled()) {
                            moex.printStackTrace();
                        }
                        LOGGER.error(moex);
                    }
                    finally {
                        if (lookupEvent != null &&
                                lookupEvent.getIntendedUse() == MOServerLookupEvent.IntendedUse.cleanUp) {
                            lookupEvent.completedUse(sreq);
                        }
                        server.unlock(sreq.getRequest(), mo);
                    }
                }
            } catch (NoSuchElementException nsex) {
                if (LOGGER.isDebugEnabled()) {
                    nsex.printStackTrace();
                }
                LOGGER.warn("Cannot find sub-request: " + nsex.getMessage());
            }
        }

        public boolean isSupported(int pduType) {
            return (pduType == AgentXPDU.AGENTX_CLEANUPSET_PDU);
        }

    }


    static class DefaultAgentXRequestFactory
            implements RequestFactory<AgentXCommandEvent<?>, AgentXResponsePDU, AgentXRequest> {

        public AgentXRequest createRequest(AgentXCommandEvent<?> initiatingEvent, CoexistenceInfo cinfo) {
            AgentXRequest request = new AgentXRequest(initiatingEvent);
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Creating AgentX request " + request +
                        " from " + initiatingEvent);
            }
            return request;
        }

    }

    class Command implements WorkerTask {

        private final AgentXCommandEvent<?> request;

        public Command(AgentXCommandEvent<?> event) {
            this.request = event;
        }

        public void run() {
            dispatchCommand(request);
        }

        public void terminate() {
        }

        public void join() throws InterruptedException {
        }

        public void interrupt() {
        }

    }


    static protected class RequestID implements Comparable<RequestID> {
        private final Address masterAddress;
        private final int sessionID;
        private final int transactionID;

        public RequestID(Address masterAddress, int sessionID, int transactionID) {
            this.masterAddress = masterAddress;
            this.sessionID = sessionID;
            this.transactionID = transactionID;
        }

        public int compareTo(RequestID other) {
            ByteBuffer ma = ByteBuffer.wrap(masterAddress.toByteArray());
            ByteBuffer oa = ByteBuffer.wrap(other.masterAddress.toByteArray());
            int c = ma.compareTo(oa);
            if (c == 0) {
                c = sessionID - other.sessionID;
                if (c == 0) {
                    c = transactionID - other.transactionID;
                }
            }
            return c;
        }

        public boolean equals(Object obj) {
            return obj instanceof RequestID && (compareTo((RequestID) obj) == 0);
        }

        public int hashCode() {
            return transactionID;
        }

    }

    class PingTask extends TimerTask {

        public void run() {
            List<AgentXSession<?>> l = new LinkedList<>(sessions.values());
            for (AgentXSession<?> session : l) {
                if (!session.isClosed()) {
                    for (OctetString context : getContexts()) {
                        ping(session, context);
                    }
                }
            }
        }

        private <A extends Address> void ping(AgentXSession<A> session, OctetString context) {
            AgentXPingPDU ping = new AgentXPingPDU(context);
            ping.setSessionAttributes(session);
            ping.setTransactionID(getNextTransactionID());
            PingEvent<A> pingEvent;
            try {
                AgentXPeer<A> agentXPeer = session.getPeer();
                AgentXResponseEvent<?> resp = agentX.send(ping, new AgentXTarget<A>(agentXPeer.getAddress(),
                        session.getTimeout()*1000), agentXPeer.getTransport());
                pingEvent = new PingEvent<>(this, session, resp.getResponse());
            } catch (IOException ex) {
                pingEvent = new PingEvent<>(this, session, ex);
            }
            firePinged(pingEvent);
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Fired ping event " + pingEvent);
            }
            if (pingEvent.isCloseSession() || pingEvent.isResetSession()) {
                try {
                    closeSession(session.getSessionID(),
                            AgentXProtocol.REASON_TIMEOUTS);
                    if (pingEvent.isResetSession()) {
                        reopenSession(session);
                    }
                } catch (IOException ex1) {
                    LOGGER.warn("IOException while resetting AgentX session in PingTask:" +
                            ex1.getMessage());
                }
            }
        }

        /**
         * Reopens a closed session.
         *
         * @param session
         *         a closed AgentXSession instance.
         *
         * @return {@link AgentXProtocol#AGENTX_SUCCESS} if the session could be opened
         * successfully. Otherwise, the AgentX error status is returned.
         * @throws IOException
         *         if the session cannot be reopened due to an IO exception.
         */
        public <A extends Address> int reopenSession(AgentXSession<A> session) throws IOException {
            return openSession(session.getPeer().getTransport(), session.getPeer().getAddress(), session);
        }

    }

    /**
     * Unlock the provided {@link ManagedObject} if the also provided {@link LockRequest} indicates
     * that the managed object was locked by a preceding {@link MOServer#lookup(MOQuery, LockRequest)} operation.
     *
     * @param server
     *         a MOServer that put the lock.
     * @param mo
     *         the possibly locked managed object.
     * @param lockRequest
     *         the lock request with the status of the (potentially acquired) lock.
     *
     * @since 4.0.0
     */
    private static void unlockManagedObjectIfLockedByLookup(MOServer server, ManagedObject<?> mo, LockRequest lockRequest) {
        switch (lockRequest.getLockRequestStatus()) {
            case locked:
            case lockedAfterTimeout:
                server.unlock(lockRequest.getLockOwner(), mo);
        }
    }

}
