# SessionManager.java

    /*
    ** Module   : SessionManager.java
    ** Abstract : Abstract base class for session management
    **
    ** Copyright (c) 2007-2025, Golden Code Development Corporation.
    **
    ** -#- -I- --Date-- --JPRM-- ---------------------------Description-------------------------------
    ** 001 ECF 20071101   @35865 Created initial version. Abstract base class
    **                           for session management. Used code from other
    **                           classes as a base.
    ** 002 ECF 20071121   @35937 Fixed initialization sequence. initialize()
    **                           now resets basic state, then calls init(),
    **                           which can be overridden by subclasses.
    ** 003 ECF 20071127   @36047 Added getSecurityPlugin() method. Added
    **                           certAlias parameter to a connectVirtual()
    **                           variant.
    ** 004 ECF 20080213   @37066 Added isLocalNode(). Reports whether a given
    **                           InetSocketAddress represents the local P2J
    **                           node.
    ** 005 GES 20081024   @40331 Added support for establishing a session
    **                           via an insecure socket (non-SSL).
    ** 006 SIY 20090512   @42141 Added support for waiting for shutdown.
    ** 007 GES 20090816   @43639 Minor signature change and formatting cleanup.
    ** 008 CA  20090825   @43756 Fixed a deadlock between Queue and SessionManager.
    **                           This was discovered during remote DB connection
    **                           closing.
    ** 009 CA  20090901   @43816 Renamed sessionsByContext. Created a map to hold
    **                           control instances for each context. Now each of
    **                           the per-context map holds the virtual control/
    **                           session for the virtual context ID. The interrupt
    **                           handler is no longer kept per each control
    **                           instance; instead is set for the SessionManager
    **                           singleton, and will be the same for the entire
    **                           server.
    ** 010 CA  20090908   @43836 Fixes a NPE in processEcho (introduced by H009
    **                           @43816).
    ** 011 CA  20090917   @43930 Fix abnormal connection end. In endSession, the TM
    **                           will be notified to end the Conversation thread, as
    **                           the session is being closed.
    ** 012 CA  20090918   @43934 H011 (@43930) should be applied only if the server
    **                           is not shutting down as the TM call isn't executed
    **                           within the correct context.
    ** 013 CA  20090918   @43948 Another case of H011 regression - TM.abortProcessing
    **                           should be called only for DirectSession's.
    ** 014 GES 20091020   @44176 Make sure to interrupt long running server threads
    **                           when the session ends. Previously this was not
    **                           done if the queue was already dead, but that is
    **                           too conservative since it will never get called
    **                           due to how the protocol closes down.
    ** 015 GES 20091103   @44305 Added logging of some core connection parms.
    ** 016 CA  20091203   @44473 Virtual sessions must not interrupt the
    **                           Conversation thread while shuting down.
    ** 017 CA  20100329   @44783 Added terminateVirtual, which allows sub-classes
    **                           to provide virtual session termination code.
    ** 018 CA  20110105          Changes to allow virtual sessions from leaf nodes
    ** 019 CA  20110131          Changes to allow the direct sessions to authenti-
    **                           cate using different credentials. Also, this way
    **                           sessions can be created to different router nodes.
    ** 020 CA  20110202          Fixed a bug in H019, related to server-to-server
    **                           connections.
    ** 021 LMR 20101210          Changed createQueue() to default
    **                           net/connection/secure to false.
    ** 022 CA  20111209          Added getSessionContexts, to get the all active
    **                           sessions which have invoked APIs in the given jar.
    **                           Also, notify the ExportTracker when a session is
    **                           finished, to be removed from all tracking data.
    ** 023 CS  20130204          Added java.net.Socket import to disambiguate from
    **                           com.goldencode.p2j.util.Socket class
    ** 024 CA  20130529          Added appserver support.
    ** 025 CA  20130628          Context-local data for this class must not be reset by the agent.
    ** 026 CA  20130705          Exposed the bootstrap configuration.
    ** 027 CA  20130822          Fixed the net protocol to support async requests.
    ** 028 CA  20140206          Added a way of notifying when the server sockets are ready (needed by
    **                           the scheduler, to know when to start launching processes).
    ** 029 CA  20140513          Added a weight for the context-local var, to ensure predetermined
    **                           order during context reset.
    ** 030 OM  20160128          Added javadoc. Code cleanup & upgrade to Java7.
    ** 031 EVL 20160223          Javadoc fixes to make compatible with Oracle Java 8 for Solaris 10.
    ** 032 CA  20170228          Added executeInContext.
    ** 033 CA  20170427          When a queue is stopped, context is set explicitly for each session
    **                           associated with this queue, when notifications are sent.
    ** 034 CA  20180217          Virtual sessions can't cleanup their context concurrently, as they
    **                           might perform remote calls (for e.g. database disconnection) - thus,
    **                           they need to use a common lock.
    ** 035 GES 20190322          Added isConversationThread() helper.
    ** 036 IAS 20200722          Refactored to the new NetSocket API, added failed handshake handling.
    **     CA  20201003          Use an identity HashSet for the async threads.
    **     IAS 20210325          Added NIO configuration via BootstrapConfig
    **     IAS 20210505          Changed exception thrown on failed shutdown.
    **     IAS 20210827          Added allowed ciphers' filtering
    **     IAS 20220624          Added support for the configurable JCE/JSSE provider
    **     TJD 20220713          Synchronization and locking optimizations
    **     TJD 20220504          Upgrade do Java 11 minor changes
    ** 037 GBB 20230512          Logging methods replaced by CentralLogger/ConversionStatus.
    ** 038 GBB 20230825          checkCallerAbort moved to SecurityUtil & calls updated.
    ** 039 TJD 20240123          Java 17 compatibility updates
    ** 040 ICP 20240704          Corrected SecurityUtil.checkCallerAbort call in getSessionContexts.
    ** 041 GBB 20240709          Hard-coded config name replaced by ConfigItem constant.
    ** 042 CA  20240910          'net:socket:nio' is enabled by default only for Java 8.
    ** 043 TJD 20231012          SessionManager.lock access optimizations
    ** 044 RNC 20250127          Added default methods 'setSessionLock' and 'getSessionLock'.
    ** 045 AL2 20250523          'net:socket:nio' is disabled by default.
    */

    /*
    ** This program is free software: you can redistribute it and/or modify
    ** it under the terms of the GNU Affero General Public License as
    ** published by the Free Software Foundation, either version 3 of the
    ** License, or (at your option) any later version.
    **
    ** 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 Affero General Public License for more details.
    **
    ** You may find a copy of the GNU Affero GPL version 3 at the following
    ** location: https://www.gnu.org/licenses/agpl-3.0.en.html
    **
    ** Additional terms under GNU Affero GPL version 3 section 7:
    **
    **   Under Section 7 of the GNU Affero GPL version 3, the following additional
    **   terms apply to the works covered under the License.  These additional terms
    **   are non-permissive additional terms allowed under Section 7 of the GNU
    **   Affero GPL version 3 and may not be removed by you.
    **
    **   0. Attribution Requirement.
    **
    **     You must preserve all legal notices or author attributions in the covered
    **     work or Appropriate Legal Notices displayed by works containing the covered
    **     work.  You may not remove from the covered work any author or developer
    **     credit already included within the covered work.
    **
    **   1. No License To Use Trademarks.
    **
    **     This license does not grant any license or rights to use the trademarks
    **     Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
    **     of Golden Code Development Corporation. You are not authorized to use the
    **     name Golden Code, FWD, or the names of any author or contributor, for
    **     publicity purposes without written authorization.
    **
    **   2. No Misrepresentation of Affiliation.
    **
    **     You may not represent yourself as Golden Code Development Corporation or FWD.
    **
    **     You may not represent yourself for publicity purposes as associated with
    **     Golden Code Development Corporation, FWD, or any author or contributor to
    **     the covered work, without written authorization.
    **
    **   3. No Misrepresentation of Source or Origin.
    **
    **     You may not represent the covered work as solely your work.  All modified
    **     versions of the covered work must be marked in a reasonable way to make it
    **     clear that the modified work is not originating from Golden Code Development
    **     Corporation or FWD.  All modified versions must contain the notices of
    **     attribution required in this license.
    */

    package com.goldencode.p2j.net;

    import java.io.*;
    import java.net.*;
    import java.util.*;
    import java.util.concurrent.*;
    import java.util.logging.*;
    import java.util.stream.*;

    import javax.net.ssl.*;
    import com.goldencode.p2j.cfg.*;
    import com.goldencode.p2j.security.*;
    import com.goldencode.p2j.security.SecurityManager;
    import com.goldencode.p2j.util.*;
    import com.goldencode.p2j.util.logging.*;

    import java.net.Socket;
    import java.nio.channels.*;
    import java.security.*;

    /**
     * This class is the central focal point of all queue and session management
     * for the net package.  It provides the public API for application code to
     * manage sessions and, indirectly, their underlying message transport queues.
     * The singleton instance of this class is created by the {@link
     * SessionManagerFactory} and is retrieved with the static {@link #get()}
     * method.
     * <p>
     * This class may not be instantiated directly.  Much of the work is performed
     * by concrete subclasses, though common functionality is implemented here.
     * The API, while asymmetric in places depending upon whether the current node
     * is a leaf or router, is provided here in its entirety to simplify use of
     * this class from both within and outside of, the net package.  Leaf methods
     * that are inappropriately invoked from a router node will raise exceptions,
     * and vice versa.
     * <p>
     * The session manager is responsible for:
     * <ul>
     *   <li>Providing a mechanism for application code to bootstrap the runtime
     *       environment for either a leaf or router network node.
     *   <li>Creating, authenticating, registering and terminating direct and
     *       virtual (i.e., multiplexed) sessions with other network nodes.
     *   <li>Creating, managing, and destroying the underlying queues at the
     *       appropriate times.
     *   <li>Acting as both the requester and provider end-points in session
     *       establishment.
     *   <li>Interacting with sessions to dynamically set an appropriate security
     *       context for the current thread (router side only).
     *   <li>Reporting on the nature of the current network node (i.e., leaf vs
     *       router).
     * </ul>
     */
    public abstract class SessionManager
    {
       /** Logger */
       private static final CentralLogger LOG = CentralLogger.get(SessionManager.class);

       /** Max number of SSL handshake attempts on connect */
       protected static final int MAX_SSL_HANDSHAKE_ATTEMPTS = 3;

       /** Singleton instance of this class */
       private static SessionManager instance = null;

       /** Topology database for this network node */
       protected final Map<InetSocketAddress, Queue> topology = new HashMap<>();

       /** Use count of the known addresses for this network node */
       protected final Map<InetSocketAddress, Integer> addressUseCount= new HashMap<>();

       /** Many-to-one map of queues to all of their associated sessions */
       protected final Map<Queue, Set<BaseSession>> sessionsByQueue = new IdentityHashMap<>();

       /** One-to-one map of context IDs to sessions */
       protected final Map<Integer, BaseSession> virtualSessionsByContext = new HashMap<>();

       /** One-to-one map of virtual locks to queues */
       protected final Map<Queue, Object> virtualLocksByQueue = new ConcurrentHashMap<Queue, Object>();

       /**
        * One-to-one map of identity to queues. Used on leaf nodes and gets
        * populated after a {@link Queue} was successfully created.
        */
       protected final Map<Queue, String> identityByQueue = new HashMap<>();

       /** Map of connected app servers; used on the P2J client-side. */
       protected final Map<Queue, String> appServers = new HashMap<>();

       /** One-to-one map of multiplexed context IDs to its associated Control instance. */
       protected final Map<Integer, Control> virtualControlByContext = new HashMap<>();

       /** Active session for the current client context */
       protected final ContextLocal<BaseSession> activeSession = new ContextLocal<BaseSession>()
       {
          @Override
          public WeightFactor getWeight()
          {
             return WeightFactor.LAST;
          };

          protected boolean isResetAllowed()
          {
             return false;
          };
       };

       /** Object on which to synchronize access to session manager state */
       protected final Object lock = new Object();

       /** Security manager */
       protected SecurityManager secMgr = null;

       /** Bootstrap configuration for this network node */
       protected BootstrapConfig config = null;

       /** A set containing threads which always send async requests. */
       private final Set<Thread> asyncThreads = Collections.newSetFromMap(new IdentityHashMap<>());

       /** The handler used to process interrupt requests. */
       private InterruptHandler interruptHandler;

       /** Shutdown thread. */
       private Thread stopper = new Thread(Dispatcher.dispThreads, new Stopper(), "queueStopper");

       /** Mark start of shutdown process. */
       private volatile boolean shutdownStarted = false;

       /** Mark end of shutdown process. */
       public volatile boolean shutdownDone = false;

       /** Next ID to be assigned for a local security context. */
       private int nextContextID = (new Random()).nextInt();

       /**
        * Constructor usable only by concrete subclasses.  Enforces singleton
        * pattern.
        *
        * @throws  IllegalStateException
        *          if called more than once per JVM instance.
        */
       protected SessionManager()
       {
          synchronized (SessionManager.class)
          {
             if (instance != null)
             {
                throw new IllegalStateException(
                   "There can be only one session manager");
             }

             instance = this;

             stopper.setDaemon(true);
          }
       }

       /**
        * Report if the current threads is a conversation thread.  A conversation thread is one
        * which is dedicated to the processing for a single security session for its lifetime.
        * Knowledge of this fact can enable optimizations since the thread is known to never
        * change security contexts.
        *
        * @return   {@code true} if the current thread is a conversion thread.
        */
       public static boolean isConversationThread()
       {
          return Queue.isConversationThread();
       }

       /**
        * Method used to access the concrete, singleton session manager instance
        * for a network node.
        *
        * @return  Singleton session manager, if it has been created;  otherwise,
        *          <code>null</code>.
        */
       public synchronized static SessionManager get()
       {
          return instance;
       }

       /**
        * Register the given thread in the list of threads which always send async requests.
        *
        * @param    thread
        *           An active thread instance.
        */
       public final void registerAsyncThread(Thread thread)
       {
          synchronized (asyncThreads)
          {
             asyncThreads.add(thread);
          }
       }

       /**
        * DeRegister the given thread from the list of threads which always send async requests.
        *
        * @param    thread
        *           An active thread instance.
        */
       public final void deregisterAsyncThread(Thread thread)
       {
          synchronized (asyncThreads)
          {
             asyncThreads.remove(thread);
          }
       }

       /**
        * Check if the given thread is in the list of threads which always send async requests.
        *
        * @param    thread
        *           An active thread instance.
        *
        * @return   <code>true</code> if the thread is in the list of threads which always send async
        *           requests.
        */
       public final boolean isAsyncThread(Thread thread)
       {
          synchronized (asyncThreads)
          {
             return asyncThreads.contains(thread);
          }
       }

       /**
        * Get the interrupt handler.
        *
        * @return  The interrupt handler instance.
        */
       public final InterruptHandler getInterruptHandler()
       {
          return interruptHandler;
       }

       /**
        * Set the interrupt handler.
        *
        * @param   handler
        *          The interrupt handler instance.
        */
       public final void setInterruptHandler(InterruptHandler handler)
       {
          if (this.interruptHandler == null)
          {
             interruptHandler = handler;
          }
       }

       /**
        * Convenience method to detect whether this session manager operates on a
        * leaf node.
        *
        * @return  <code>true</code> if network node for this session manager is a
        *          leaf node;  else <code>false</code>.
        */
       public boolean isLeaf()
       {
          return false;
       }

       /**
        * Expose the {@link BootstrapConfig config} used to start the P2J Client or Server.
        *
        * @return    A copy of the {@link #config}.
        */
       public BootstrapConfig config()
       {
          return new BootstrapConfig(config);
       }

       /**
        * Convenience method to detect whether this session manager operates on a
        * router node.
        *
        * @return  <code>true</code> if network node for this session manager is a
        *          router node;  else <code>false</code>.
        */
       public boolean isRouter()
       {
          return false;
       }

       /**
        * Getter for the session lock. By default, it does nothing.
        *
        * @return  The value of the session lock.
        */
       public boolean getSessionLock()
       {
          return false;
       }

       /**
        * Setter for the session lock. By default, it does nothing.
        *
        * @param   lockStatus
        *          The new status for the lock. {@code true} if the session launching will be stopped,
        *          {@code false} otherwise
        */
       public boolean setSessionLock(boolean lockStatus)
       {
          return false;
       }

       /**
        * Get the appserver associated with this P2j client session.
        *
        * @param    session
        *           The P2J session.
        *
        * @return   The appserver's name, or <code>null</code> if the client is not started as an
        *           appserver.
        */
       public String getAppServer(Session session)
       {
          // this must be called only on the P2J client-side, where the app-server gets started.
          synchronized (appServers)
          {
             return appServers.get(((BaseSession) session).getQueue());
          }
       }

       /**
        * Convenience method to detect whether the given socket address represents
        * the current node.  This is determined by comparing the port number
        * with this node's port number, and by comparing the given host address
        * with each address available on every network interface for the local
        * machine.  If the port numbers match and any address matches the given
        * host address, the given address is considered to represent the current
        * network node.
        * <p>
        * This implementation throws an exception.  It must be overridden to
        * provide the functionality described above.
        *
        * @param   address
        *          Socket address to check.
        *
        * @return  <code>true</code> if an address match is detected, else
        *          <code>false</code>.
        *
        * @throws  UnsupportedOperationException
        *          always.
        */
       public boolean isLocalNode(InetSocketAddress address)
       {
          throw new UnsupportedOperationException();
       }

       /**
        * Called by a requester to authenticate a virtual session.  Invoked via
        * remote call on the provider side of a connection to establish the
        * provider's side of a virtual session.
        * <p>
        * This implementation throws an exception.  It must be overridden to
        * provide the functionality described above.
        *
        * @param   acctID
        *          Account identifier for the requesting party.
        * @param   contextID
        *          Value which uniquely identifies the security context of the
        *          requesting party.  This will be used in communications from
        *          the provider side of the connection.
        * @param   queue
        *          Message transport queue for the provider side of the
        *          connection.
        *
        * @return  Generated security context ID.  This uniquely identifies the
        *          newly created provider-side security context for the virtual
        *          session.  It is used by the requester to multiplex
        *          communications for this session over a shared queue.  This ID
        *          will never be 0, which is reserved for direct sessions.
        *
        * @throws  UnsupportedOperationException
        *          always.
        * @throws  IllegalAccessException
        *          if authentication is not successful.
        */
       public int authenticateRemote(String acctID, int contextID, Queue queue)
       throws IllegalAccessException
       {
          throw new UnsupportedOperationException();
       }

       /**
        * Called by a requester to terminate a virtual session.  Invoked via
        * remote call on the provider side of a connection to terminate the
        * provider's side of a virtual session.
        * <p>
        * This implementation throws an exception.  It must be overridden to
        * provide the functionality described above.
        *
        * @param   contextID
        *          Value which uniquely identifies the security context of the
        *          requesting party.
        *
        * @throws  UnsupportedOperationException
        *          always.
        */
       public void terminateRemote(int contextID)
       {
          throw new UnsupportedOperationException();
       }

       /**
        * Called by a requester to initiate virtual session termination. This is
        * invoked on the requester side of a connection to initiate termination of
        * both the requester and provider sides of a virtual session.
        *
        * @param   session
        *          The session which needs to be terminated.
        */
       // Removed synchronized keyword as it does not need to be synchronized as all variables are synchronized
       // before access
       public void terminateVirtual(BaseSession session)
       {
          Queue queue = session.getQueue();

          if (!queue.isRunning())
          {
             // queue already down, nothing can be done
             return;
          }

          Object queueLock = null;

          // this lock ensures synchronization per each queue
          queueLock = virtualLocksByQueue.get(queue);

          // if lock was removed do nothing
          if (queueLock == null)
          {
             return;
          }
          synchronized (queueLock)
          {
             queueLock = virtualLocksByQueue.get(queue);
             // if lock was removed in meantime do nothing
             if (queueLock == null)
             {
                return;
             }
             callTerminateRemote(session);
             endSession(session, false, true);

             if (!queue.isRunning())
             {
                virtualLocksByQueue.remove(queue);
             }
          }
       }

       /**
        * Make a synchronous method call to a remote router node to terminate the
        * remote side of this virtual session.  This must be done while the
        * session's queue is still active, in the event this session's termination
        * is to be followed by the termination of its queue.
        * @param session
        *        session the be terminated
        */
       private void callTerminateRemote(BaseSession session)
       {
          if (!session.isRunning())
          {
             // Can't call peer side if queue is already down.
             return;
          }

          Message message = new Message(MessageTypes.REQUEST_SYNCH);
          RoutingKey key = new RoutingKey(RoutingKey.TERMINATE_REMOTE);
          key.setSourceAddress(session.getNodeAddress());
          key.setDestinationAddress(session.getRemoteAddress());
          message.setKey(key);

          int contextID = session.getContextID();
          message.setPayload(new Object[] { Integer.valueOf(contextID) });

          try
          {
             session.transact(message, 0);
          }

          catch (ApplicationRequestedStop exc)
          {
             if (LOG.isLoggable(Level.FINE))
             {
                LOG.logp(Level.FINE,
                         "SessionManager.callTerminateRemote()",
                         "",
                         exc.getMessage());
             }
          }

          catch (Exception exc)
          {
             // check if the cause is a InterruptedException. in this case, do
             // not log it as a warning
             Level logLvl = Level.WARNING;
             String msg = "Error terminating virtual session peer.";

             Throwable cause = exc;
             while (cause != null)
             {
                if (cause instanceof InterruptedException)
                {
                   logLvl = Level.FINE;
                   break;
                }
                if (cause instanceof SilentUnwindException)
                {
                   msg = "SessionManager.callTerminateRemote()";
                   logLvl = Level.FINE;
                   break;
                }

                cause = cause.getCause();
             }

             if (LOG.isLoggable(logLvl))
             {
                LOG.log(logLvl, msg, exc);
             }
          }
       }

       /**
        * Establish a direct connection to a network node using the bootstrap
        * configuration stored during construction.
        * <p>
        * This implementation calls {@link #connectDirect(BootstrapConfig,
        * InterruptHandler, SessionListener)}, with a <code>null</code> bootstrap
        * config. So, the {@link #config} instance received during this manager's
        * {@link SessionManagerFactory#createLeafNode(BootstrapConfig) creation}
        * will be used instead.
        *
        * @param   interruptHandler
        *          An object that is notified when the current thread is
        *          interrupted.
        * @param   notify
        *          Session listener to be notified of session events.
        *
        * @return  Session established after the connection has been made.
        *
        * @throws  UnsupportedOperationException
        *          always.
        * @throws  Exception
        *          not thrown.
        */
       public Session connectDirect(InterruptHandler interruptHandler,
                                    SessionListener notify)
       throws Exception
       {
          return connectDirect(null, interruptHandler, notify);
       }

       /**
        * Establish a direct connection to a network node using the specified
        * bootstrap configuration.
        * <p>
        * This implementation throws an exception.  It must be overridden to
        * provide the functionality described above.
        *
        * @param   config
        *          Bootstrap configuration information which contains at least
        *          enough information (host and port) to make a direct connection
        *          from this node to another.
        * @param   interruptHandler
        *          An object that is notified when the current thread is
        *          interrupted.
        * @param   notify
        *          Session listener to be notified of session events.
        *
        * @return  Session established after the connection has been made.
        *
        * @throws  UnsupportedOperationException
        *          always.
        * @throws  Exception
        *          not thrown.
        */
       public Session connectDirect(BootstrapConfig config,
                                    InterruptHandler interruptHandler,
                                    SessionListener notify)
       throws Exception
       {
          throw new UnsupportedOperationException();
       }

       /**
        * Establish a virtual (multiplexed) connection to a network node at the
        * given address.
        * <p>
        * This implementation throws an exception.  It must be overridden to
        * provide the functionality described above.
        *
        * @param   address
        *          Socket address of remote network node.
        * @param   certAlias
        *          Alias of certificate to be read from trust store to enable
        *          certificate-based, server-to-server authentication.  This
        *          should match the server ID of the remote node.
        * @param   notify
        *          Session listener to be notified of session events.
        *
        * @return  Session established after the connection has been made.
        *
        * @throws  UnsupportedOperationException
        *          always.
        * @throws  Exception
        *          not thrown.
        */
       public Session connectVirtual(InetSocketAddress address,
                                     String certAlias,
                                     SessionListener notify)
       throws Exception
       {
          return connectVirtual(address, certAlias, notify, null);
       }

       /**
        * Establish a virtual (multiplexed) connection to a network node at the
        * given address.
        * <p>
        * This implementation throws an exception.  It must be overridden to
        * provide the functionality described above.
        *
        * @param   address
        *          Socket address of remote network node.
        * @param   certAlias
        *          Alias of certificate to be read from trust store to enable
        *          certificate-based, server-to-server authentication.  This
        *          should match the server ID of the remote node.
        * @param   notify
        *          Session listener to be notified of session events.
        * @param   account
        *          Explicit account to authenticate on the remote side. May be <code>null</code>.
        *
        * @return  Session established after the connection has been made.
        *
        * @throws  UnsupportedOperationException
        *          always.
        * @throws  Exception
        *          not thrown.
        */
       public Session connectVirtual(InetSocketAddress address,
                                     String            certAlias,
                                     SessionListener   notify,
                                     String            account)
       throws Exception
       {
          throw new UnsupportedOperationException();
       }

       /**
        * Establish a virtual (multiplexed) connection to a network node at the
        * address specified by the given bootstrap configuration.
        * <p>
        * This implementation throws an exception.  It must be overridden to
        * provide the functionality described above.
        *
        * @param   config
        *          Bootstrap configuration information which contains at least
        *          enough information (host and port) to make a virtual connection
        *          from this node to another.
        * @param   notify
        *          Session listener to be notified of session events.
        *
        * @return  Session established after the connection has been made.
        *
        * @throws  UnsupportedOperationException
        *          always.
        * @throws  Exception
        *          not thrown.
        */
       public Session connectVirtual(BootstrapConfig config,
                                     SessionListener notify)
       throws Exception
       {
          throw new UnsupportedOperationException();
       }

       /**
        * Listen for incoming socket connections, authenticate each new
        * connection, and establish a new queue for each authenticated connection.
        * <p>
        * This implementation throws an exception.  It must be overridden to
        * provide the functionality described above.
        *
        * @param   notify
        *          Session listener to be notified of session events.
        *
        * @throws  UnsupportedOperationException
        *          always.
        * @throws  Exception
        *          not thrown.
        */
       public void listen(SessionListener notify)
       throws Exception
       {
          throw new UnsupportedOperationException();
       }

       /**
        * Get the active session, if any.  This is the last session whose security
        * context was set.
        *
        * @return  Active session, or <code>null</code>.
        */
       public Session getSession()
       {
          return activeSession.get();
       }

       /**
        * Check if the current session is a {@link VirtualSession}.
        *
        * @return    See above.
        */
       public boolean isVirtualSession()
       {
          return activeSession.get() instanceof VirtualSession;
       }

       /**
        * Get the contexts for all active sessions which have invoked APIs in the
        * given jar.  If the jar name is <code>null</code>, list all active
        * sessions.
        * <p>
        * <i><u>Restricted use</u></i>. This method checks the caller to be
        * <code>com.goldencode.p2j.security.SecurityManager.getActiveSessions</code>.
        *
        * @param   jar
        *          The jar name
        * @param   iface
        *          The interface for which we need the sessions;
        *          if <code>null</code>, search through the entire jar.
        *
        * @return  See above.
        */
       public Object[] getSessionContexts(String jar, String iface)
       {
          try
          {
             SecurityUtil.checkCallerAbort(
                new String[]
                {
                   "com.goldencode.p2j.security.SecurityManager.getActiveSessions",
                   "com.goldencode.p2j.security.SecurityAdmin.getActiveSessions"
                },
                "com.goldencode.p2j.net.SessionManager"
             );
          }
          catch (RestrictedUseException e)
          {
             if (LOG.isLoggable(Level.SEVERE))
             {
                LOG.log(Level.SEVERE,
                        "Unauthorized call of SessionManager.getSessionContexts",
                        e);
             }

             return new Object[0];
          }

          Set<Session> sessions = ExportTracker.listSessions(jar, iface);
          List<Object> contexts = new ArrayList<>();
          for(Session sess : sessions)
          {
             BaseSession base = (BaseSession) sess;
             contexts.add(base.getContext());
          }

          return contexts.toArray();
       }

       /**
        * Terminate all active sessions which have invoked the given APIs in the
        * given jar. If the jar name is <code>null</code>, terminate all active
        * tracked sessions.
        * This method guarantees that all session termination listeners have been
        * executed.
        * <p>
        * <i><u>Restricted use</u></i>. This method checks the caller to be
        * <code>com.goldencode.p2j.security.SecurityManager.terminateSessions</code>.
        *
        * @param  jar
        *         The jar name; if <code>null</code>, terminate all active tracked
        *         sessions.
        * @param  iface
        *         The interface for which we need to terminate the sessions; if
        *         <code>null</code>, terminate all sessions for the given jar.
        *
        * @throws RestrictedUseException
        *         If called improperly.
        */
       public void terminateSessions(String jar, String iface)
       throws RestrictedUseException
       {
          SecurityUtil.checkCallerAbort(
                 "com.goldencode.p2j.security.SecurityManager.terminateSessions",
                 "com.goldencode.p2j.net.SessionManager");

          Set<Session> sessions = ExportTracker.listSessions(jar, iface);

          for (Session session : sessions)
          {
             session.terminate();
          }
       }

       /**
        * Get the local node address of the active session, if any.  This is the
        * last session whose security context was set.
        *
        * @return  Active session's local node address, or 0 if no active session
        *          exists.
        */
       public int getNodeAddress()
       {
          Session session = getSession();

          return (session != null ? session.getNodeAddress() : 0);
       }

       /**
        * Get the remote node address of the active session, if any.  This is the
        * last session whose security context was set.
        *
        * @return  Active session's remote node address, or 0 if no active session
        *          exists.
        */
       public int getRemoteAddress()
       {
          Session session = getSession();

          return (session != null ? session.getRemoteAddress() : 0);
       }

       /**
        * Terminate ALL sessions managed by this session manager and shut down the
        * network node itself.  This involves terminating the connection listening
        * loop (router node only), the node's primary transport queue, and the
        * router queue (router node only).
        *
        * @throws  AccessControlException
        *          if the caller's security context does not include sufficient
        *          rights to perform the shutdown.
        */
       public void shutdown()
       throws SecurityException
       {
          // Fail the call if not sufficient access rights.
          if (!secMgr.checkShutdownAccess())
          {
             throw new SecurityException("Insufficient access rights for SHUTDOWN.");
          }

          synchronized (stopper)
          {
             if (!shutdownStarted)
             {
                shutdownStarted = true;
                stopper.start();
             }
          }
       }

       /**
        * Block the calling thread until the network portions of the server are fully initialized.
        * At that point the server is ready to receive socket connections.
        * @throws InterruptedException
        *         if the current thread was interrupted
        */
       public void waitUntilReady()
       throws InterruptedException
       {
          throw new UnsupportedOperationException();
       }

       /**
        * Wait for end of shutdown thread. If shutdown thread was not started, it
        * is started immediately.
        *
        * @throws   AccessControlException
        *           if the caller's security context does not include sufficient
        *           rights to perform the shutdown.
        */
       public void waitForShutdown()
       throws SecurityException
       {
          // Fail the call if not sufficient access rights.
          if (!secMgr.checkShutdownAccess())
          {
             throw new SecurityException("Insufficient access rights for SHUTDOWN.");
          }

          synchronized (stopper)
          {
             if (shutdownDone)
                return;

             if (!shutdownStarted)
             {
                shutdownStarted = true;
                stopper.start();
             }

             while (!shutdownDone)
             {
                try
                {
                   stopper.wait();
                }
                catch (Exception e)
                {

                }
             }
          }
       }

       /**
        * Return a string representation of the internal state of this object,
        * primarily for debugging purposes.
        *
        * @return  String representation of this object.
        *
        * @see java.lang.Object#toString()
        */
       public String toString()
       {
          return ("TOP: " + topology +
                  " SBQ: " + sessionsByQueue +
                  " SBC: " + virtualSessionsByContext +
                  " ACT: " + activeSession.get());
       }

       /**
        * Execute the given task in the specified session's context.
        * <p>
        * Only a thread with the server's context is allowed to execute this.
        *
        * @param    session
        *           The client session.
        * @param    task
        *           The task to execute.
        *
        * @throws   RestrictedUseException
        *           If the current context is not a server context.
        */
       public void executeInContext(Session session, Runnable task)
       throws RestrictedUseException
       {
          if (!SecurityManager.getInstance().isServerAccount())
          {
             throw new RestrictedUseException("Only server accounts are allowed to use this API!");
          }

          try
          {
             setContext(((BaseSession) session).getQueue());

             task.run();
          }
          finally
          {
             restoreContext();
          }
       }

       /**
        * Create a direct session for the given queue and register it with the
        * session manager.
        * <p>
        * This implementation throws an exception.  It must be overridden to
        * provide the functionality described above.
        *
        * @param    queue
        *           Underlying message transport for the new session.
        *
        * @return   <code>true</code> if the session initialized successfully.
        */
       protected boolean createDirectSession(Queue queue)
       {
          throw new UnsupportedOperationException();
       }

       /**
        * Create a socket and the low level message transport queue, to be
        * used to communicate with the remote node at the given socket address.
        * Authenticate this network node as a valid requester of services from the
        * remote node using the security manager.
        * <p>
        * The socket that is created is either secure or insecure, depending on
        * the value of <code>net/connection/secure</code> bootstrap configuration
        * flag.
        *
        * @param   remoteAddr
        *          Address of the socket for the remote network node.
        * @param   config
        *          BootstrapConfig which should override that of the security
        *          manager for this session.  If <code>null</code>, the security
        *          manager's default configuration is used.
        *
        * @return  The newly created queue, if authentication was successful.
        *
        * @throws  Exception
        *          if authentication fails.
        */
       protected Queue createQueue(InetSocketAddress remoteAddr, BootstrapConfig config)
       throws Exception
       {
          BootstrapConfig bc = (config == null) ? this.config : config;
          int attempts = MAX_SSL_HANDSHAKE_ATTEMPTS;
          NetSocket netSock = null;
          while(attempts-- > 0)
          {
             try
             {
                netSock = connect(remoteAddr, bc);
                break;
             }
             catch (SSLHandshakeException e)
             {
                if (attempts == 0)
                {
                   throw e;
                }
                LOG.log(Level.WARNING, String.format("SSL handshake failed, %d attempts remaining", attempts));
                continue;
             }
          }

          // force authentication before any protocol ops can start
          AuthData auth = secMgr.authenticateClient(netSock, bc);
          if (auth == null)
          {
             throw new Exception("Authentication failed");
          }

          Queue queue = new Queue(netSock);

          // testing for leaf node. The context should be irrelevant (as opposed to !isServerAccount)
          if (secMgr.isClient())
          {
             synchronized (identityByQueue)
             {
                identityByQueue.put(queue, auth.getIdentity());
             }
             if (auth.isAppServer())
             {
                synchronized (appServers)
                {
                   appServers.put(queue, auth.getAppServer());
                }
             }
          }

          return queue;
       }

       /**
        * Create the low level socket to match the session's requirements and connect it to the peer.  This
        * supports secure and insecure sockets.  For secure sockets, both NIO and the standard approach can be
        * used.
        *
        * @param    remoteAddr
        *           The address of the peer.
        * @param    bc
        *           The bootstrap configuration which defines the core connectivity.
        *
        * @return   The low level socket resource.
        *
        * @throws IOException
        * @throws SocketException
        * @throws NoSuchAlgorithmException
        * @throws KeyStoreException
        * @throws UnrecoverableKeyException
        * @throws KeyManagementException
        * @throws ConfigurationException
        */
       private NetSocket connect(InetSocketAddress remoteAddr, BootstrapConfig bc)
       throws IOException,
              SocketException,
              NoSuchAlgorithmException,
              KeyStoreException,
              UnrecoverableKeyException,
              KeyManagementException,
              ConfigurationException,
              NoSuchProviderException
       {
          // detect if we must use TLS (by default we don't use TLS)
          boolean secure = bc.getBoolean("net", "connection", "secure", false);
          boolean server = this.config.isServer();

          Socket lowSocket;
          SocketChannel channel = null;
          boolean nio = bc.getBoolean(ConfigItem.SERVER_SOCKET_NIO, false);
          if (nio)
          {
             channel = SocketChannel.open();
    //         channel.setOption(StandardSocketOptions.SO_REUSEADDR , true);
             channel.setOption(StandardSocketOptions.TCP_NODELAY , true);
             channel.connect(remoteAddr);
             lowSocket = channel.socket();
          }
          else
          {
             lowSocket = new Socket();
             lowSocket.setReuseAddress(true);
             lowSocket.bind(null);
             lowSocket.connect(remoteAddr);
          }

          Socket sock = lowSocket;

          if (LOG.isLoggable(Level.FINER))
          {
             String msg = String.format("host = %s; port = %d; secure = %b;",
                                        remoteAddr.getHostName(),
                                        remoteAddr.getPort(),
                                        secure);

             LOG.logp(Level.FINER, "SessionManager.createQueue()", "", msg);
          }

          NetSocket netSock;
          if (secure)
          {
             // if this is a server, always use the primordial configuration
             // (used to initialize the SessionManager)
             SSLContext ctx = secMgr.contextSm.getSecureSocketContext(server ? this.config : bc);
             if (nio)
             {
                SSLEngine engine = ctx.createSSLEngine();
                engine.setEnabledCipherSuites(
                         SecurityManager.getInstance().
                               filterCiphers(engine.getEnabledCipherSuites())
                );
                if (LOG.isLoggable(Level.CONFIG))
                {
                   LOG.config(
                      Arrays.stream(
                            engine.getEnabledCipherSuites()).collect(Collectors.joining(","))
                   );
                }
                engine.setUseClientMode(true);
                netSock = NetSocket.FACTORY.create(engine, channel);
             }
             else
             {
                SSLSocketFactory factory = ctx.getSocketFactory();
                String           host    = remoteAddr.getHostName();
                int              port    = remoteAddr.getPort();

                // shift the low level socket into secure mode
                sock = factory.createSocket(lowSocket, host, port, true);
                sock.setTcpNoDelay(true);
                netSock = NetSocket.FACTORY.create(sock);
             }
          }
          else
          {
             netSock = channel != null ? NetSocket.FACTORY.create(channel) :
                                         NetSocket.FACTORY.create(sock);
          }
          return netSock;
       }

       /**
        * Terminate the socket connection listening loop for a router network
        * node.
        * <p>
        * This implementation is a NOP;  it must be overridden to provide the
        * above functionality.
        */
       protected void terminateInput()
       {
       }

       /**
        * End the specified network session.  This involves:
        * <ol>
        *   <li>optionally interrupting any currently running {@link Dispatcher}
        *       or {@link Conversation} thread or session;
        *   <li>deregistering the session from the session manager's database;
        *   <li>disabling the session's queue if no other sessions are still using
        *       it;
        *   <li>optionally notifying registered session listeners of the
        *       termination event;
        *   <li>stopping the session's queue transport if no other sessions are
        *       still using it;
        *   <li>terminating the session with the security manager.
        * </ol>
        *
        * @param   session
        *          Session that is ending.
        * @param   interrupt
        *          <code>true</code> to interrupt any long running service thread
        *          asynchronously.
        * @param   notify
        *          <code>true</code> to notify registered listeners of the
        *          termination.  This should be <code>false</code> if the queue
        *          itself triggered the termination, since in that case the
        *          notification will already have occurred.
        */
       protected void endSession(BaseSession session, boolean interrupt, boolean notify)
       {
          Queue queue = session.getQueue();
          int contextID = session.getContextID();
          boolean virtual;
          // moved into synchronized section to prevent deadlocks
          synchronized(lock)
          {
             virtual = virtualControlByContext.containsKey(contextID);
          }
          // if this is a virtual session, then the Conversation thread must not be interrupted.
          if (!virtual && interrupt && isRouter())
          {
             // any further processing of the conversation thread needs to be ended
             // so we notify it. On the next TM.honorStopCondition call, TM will
             // intercept this flag and throw a SilentUnwindException.
             if (!shutdownStarted && (session instanceof DirectSession))
             {
                TransactionManager.abortProcessing();
             }

             // Interrupt any long running server thread since otherwise that
             // thread may continue for an arbitrary amount of time even though
             // the session is ended. This must happen after the abortProcessing()
             // call above so that if the interrupt() is handled by calling code,
             // the flag is set to allow the fatal nature of this condition to
             // be known (since the session must exit).
             Control.interrupt();
          }

          // Remove the session from the session manager's database.
          boolean queueInUse = deregisterSession(session);

          // Determine if session's queue needs to be stopped (i.e., it is no
          // longer needed by any other session and it is currently running).
          boolean stopQueue = (!queueInUse && queue.isRunning());

          // Disable the session's queue if necessary.
          if (stopQueue)
          {
             queue.disable(null);
          }

          // Notify listeners of the termination.
          if (notify)
          {
             session.sendTerminationEvent();
          }

          // Stop the session's queue if necessary.
          if (stopQueue)
          {
             queue.stopTransport(null);

             // remove mappings created in createQueue()
             if (secMgr.isClient())
             {
                synchronized (identityByQueue)
                {
                   identityByQueue.remove(queue);
                }
                synchronized (appServers)
                {
                   appServers.remove(queue);
                }
             }
          }

          // End the session.
          session.end();
       }

       /**
        * Register a new session with the session manager.  This updates the
        * session manager's database, which includes:
        * <ul>
        *   <li>a topology mapping of socket addresses to queues;
        *   <li>a mapping of local context IDs to sessions;
        *   <li>a mapping of queues to sets of sessions which use them.
        * </ul>
        *
        * @param   contextID
        *          An integer which uniquely identifies a virtual session on the
        *          local node.  This ID is used by the peer side of the session to
        *          multiplex communications with multiple virtual sessions over a
        *          single queue.
        * @param   remoteAddr
        *          Socket address of the network node on the remote side of the
        *          session.
        * @param   session
        *          Session being registered.
        */
       protected void registerSession(int               contextID,
                                      InetSocketAddress remoteAddr,
                                      BaseSession       session)
       {
          Queue queue = session.getQueue();

          if (remoteAddr == null)
          {
             remoteAddr = queue.getRemoteInetAddress();
          }

          synchronized (lock)
          {
             int n = addressUseCount.containsKey(remoteAddr)
                        ? addressUseCount.get(remoteAddr) : 0;
             addressUseCount.put(remoteAddr, ++n);
             topology.put(remoteAddr, queue);

             if (contextID != 0)
             {
                virtualSessionsByContext.put(contextID, session);
             }

             // virtual sessions require to cleanup on the same lock... because they use the same
             // queue for transport, and might require to perform disconnect - which can't be done
             // concurrently
             secMgr.setCleanupLock(session.getContext(), queue.getCleanupLock());

             Set<BaseSession> set = sessionsByQueue.get(queue);
             if (set == null)
             {
                set = new HashSet<BaseSession>();
                sessionsByQueue.put(queue, set);
             }
             set.add(session);
          }

          // notify the export tracker when the session ends
          session.addSessionListener(ExportTracker.getTracker());
       }

       /**
        * Deregister a session with the session manager.  This updates the session
        * manager's database, which includes:
        * <ul>
        *   <li>a topology mapping of socket addresses to queues;
        *   <li>a mapping of local context IDs to sessions;
        *   <li>a mapping of queues to sets of sessions which use them.
        * </ul>
        * <p>
        * If the session being deregistered is the last session to use its queue,
        * this fact is indicated by the method's return value.
        * <p>
        * This method is tolerant of being called multiple times for the same
        * session instance, since there are several reasons this method may be
        * called at session termination:
        * <ul>
        *   <li>explicit termination by a call to {@link Session#terminate()};
        *   <li>the queue going down, due either to a connectivity error, or to
        *       the connection being disrupted by the termination (normal or
        *       abnormal) of the queue's peer.
        * </ul>
        *
        * @param   session
        *          Session being deregistered.
        *
        * @return  <code>true</code> if the session's underlying queue is still
        *          needed by other sessions;  <code>false</code> if this was the
        *          last session using its queue, and the queue can now be stopped.
        */
       protected boolean deregisterSession(BaseSession session)
       {
          assert (session != null);

          boolean queueInUse = true;
          Queue queue = session.getQueue();

          synchronized (queue)
          {
             synchronized (lock)
             {
                int contextID = session.getContextID();

                if (contextID != 0)
                {
                   virtualSessionsByContext.remove(contextID);
                   virtualControlByContext.remove(contextID);
                }

                Set<BaseSession> set = sessionsByQueue.get(queue);
                if (set == null)
                {
                   // This session already was deregistered, as the queue is no
                   // longer in use.
                   return false;
                }

                set.remove(session);

                InetSocketAddress remoteAddr = queue.getRemoteInetAddress();
                int n = addressUseCount.get(remoteAddr) - 1;
                if (n == 0)
                {
                   topology.remove(remoteAddr);
                   addressUseCount.remove(remoteAddr);
                }
                else
                {
                   addressUseCount.put(remoteAddr, n);
                }

                if (set.isEmpty())
                {
                   sessionsByQueue.remove(queue);
                   queueInUse = false;
                }
             }
          }

          return queueInUse;
       }

       /**
        * Set the security context for the current thread, using the session
        * identified by the given routing key or transport queue.
        * <p>
        * This implementation is a NOP;  it must be overridden to provide the
        * above functionality.
        *
        * @param   key
        *          Routing key for an incoming message.
        * @param   queue
        *          Message transport queue.
        */
       protected void setContext(RoutingKey key, Queue queue)
       {
       }

       /**
        * Set the security context for the current thread, using the session
        * identified by the given queue.  It is assumed that there is one and only
        * one session using the queue.
        * <p>
        * This implementation is a NOP;  it must be overridden to provide the
        * above functionality.
        *
        * @param   queue
        *          Queue which uniquely identifies the session whose security
        *          context is to be used.
        */
       protected void setContext(Queue queue)
       {
       }

       /**
        * Set the security context for the current thread, using the security
        * context stored in the given session.  Mark this session as the active
        * session for this context.
        *
        * @param   session
        *          Session whose security context is to be associated with the
        *          current thread.
        */
       protected void setContext(BaseSession session)
       {
       }

       /**
        * Restore the current thread's security context to whatever it was before
        * it was last explicitly set.
        * <p>
        * This implementation is a NOP;  it must be overridden to provide the
        * above functionality.
        */
       protected void restoreContext()
       {
       }

       /**
        * Initialize security context switching for the calling thread.
        * <p>
        * This implementation is a NOP;  it must be overridden to provide the
        * above functionality.
        */
       protected void setInitialContext()
       {
       }

       /**
        * Release the initial security context assigned to the calling thread.
        * <p>
        * This implementation is a NOP;  it must be overridden to provide the
        * above functionality.
        */
       protected void dropInitialContext()
       {
       }

       /**
        * Initialize the session manager beyond what is done by the {@link
        * #initialize(BootstrapConfig)} method.
        * <p>
        * This implementation does nothing.  It should be overridden by subclasses
        * with special initialization requirements.
        *
        * @throws  ConfigurationException
        *          if there is any configuration error.
        */
       protected void init()
       throws ConfigurationException
       {
       }

       /**
        * Generate a security context ID which uniquely identifies the security
        * context of a virtual session.  This ID will never be 0, which is
        * reserved for direct connections.
        *
        * @return  Unique security context ID.
        */
       protected int nextContextID()
       {
          int id;
          synchronized (lock)
          {
             // Advance to the next local context ID not already in use;  note
             // that 0 is reserved for direct sessions.
             id = nextContextID;
             while (++id == 0 || virtualSessionsByContext.containsKey(id));
             nextContextID = id;

             // Store placeholder to reserve this identifier.
             virtualSessionsByContext.put(id, null);
          }

          return id;
       }

       /**
        * Make a synchronous method call to a remote router node to authenticate
        * the specified account and retrieve the remote node's context ID for the
        * newly authenticated, virtual session.  Also provide this node's context
        * ID for its side of the virtual session, so that the remote node can
        * initiate transactions with the proper virtual session on this side.
        *
        * @param   acctID
        *          User or process account identifier;  must be known to the
        *          remote node.
        * @param   contextID
        *          Value which uniquely identifies the security context of this
        *          side of the virtual session.
        * @param   queue
        *          Queue used to transact the authentication method call.
        *
        * @return  Value which uniquely identifies the security context of the
        *          remote node's side of the virtual session.
        *
        * @throws  Exception
        *          if an exception is thrown by the remote method invoked.
        */
       protected int callAuthenticateRemote(String acctID,
                                            int contextID,
                                            Queue queue)
       throws Exception
       {
          Message message = new Message(MessageTypes.REQUEST_SYNCH);

          RoutingKey key = new RoutingKey(RoutingKey.AUTHENTICATE_REMOTE);
          key.setSourceAddress(queue.getNodeAddress());
          key.setDestinationAddress(queue.getRemoteAddress());
          message.setKey(key);

          Object[] payload = new Object[]
          {
             acctID,
             contextID
          };
          message.setPayload(payload);

          Integer reply = (Integer) queue.transact(message, 0);

          return reply;
       }

       /**
        * Initialize the session manager.  This method is called by the {@link
        * SessionManagerFactory} after the session manager instance has been
        * constructed or retrieved for reuse.
        *
        * @param   config
        *          Bootstrap configuration for this network node.
        *
        * @throws  ConfigurationException
        *          if there is any configuration error.
        */
       final void initialize(BootstrapConfig config)
       throws ConfigurationException
       {
          if (config == null)
          {
             throw new ConfigurationException(
                "Bootstrap configuration is required");
          }

          this.config = config;

          secMgr = SecurityManager.getInstance();
          assert (secMgr != null);

          synchronized (lock)
          {
             topology.clear();
             addressUseCount.clear();
             sessionsByQueue.clear();
             virtualSessionsByContext.clear();
             virtualControlByContext.clear();
             activeSession.set(null);

             // Give subclasses a chance to do additional initialization.
             init();
          }
       }

       /**
        * Get the security plugin which is responsible for the resources of the
        * specified type.  Delegates to the <code>SecurityManager</code>.  If
        * the current node is a leaf node, <code>null</code> is returned.
        * <p>
        * Important Note:  a security context should be set before this method is
        * invoked.  Otherwise, <code>null</code> will be returned.
        *
        * @param   resourceTypeName
        *          Type of resource managed by the security plugin.
        *
        * @return  <code>AbstractResource</code> object, or <code>null</code>.
        */
       AbstractResource getSecurityPlugin(String resourceTypeName)
       {
          return secMgr.getPluginInstance(resourceTypeName);
       }

       /**
        * For each session associated with the given queue, notify all of the
        * session's session listeners of the termination event.  At the time this
        * method is invoked, the queue has been disabled for new messages, but the
        * underlying transport mechanism is still active.
        *
        * @param   queue
        *          Queue which is terminating.
        *
        * @see     #queueStopped(Queue)
        */
       void notifyTermination(Queue queue)
       {
          Set<BaseSession> set = null;
          synchronized (lock)
          {
             set = sessionsByQueue.get(queue);
          }

          if (set != null)
          {
             for (BaseSession next : set)
             {
                setContext(next);

                try
                {
                   next.sendTerminationEvent();
                }
                finally
                {
                   restoreContext();
                }
             }
          }

          // queue is terminating, remove the virtual lock (if any)
          virtualLocksByQueue.remove(queue);
       }

       /**
        * End all sessions associated with the given queue, since the queue has
        * been stopped.  By the time this is called, the low level message
        * transport is no longer running.  This method is about cleaning up higher
        * level session resources, including security contexts.  All session
        * listeners already have been notified of the termination.
        *
        * @param   queue
        *          Queue which has been stopped.
        *
        * @see     #notifyTermination(Queue)
        */
       void queueStopped(Queue queue)
       {
          synchronized (queue)
          {
             Set<BaseSession> set = null;
             synchronized (lock)
             {
                // Iterate through all sessions for the given queue, stopping each
                // one.  We make a copy of the set of sessions to avoid a concurrent
                // modification exception when the session is deregistered from the
                // session manager's database.
                set = sessionsByQueue.get(queue);

                if (set == null)
                {
                   return;
                }
             }
             // moved out internal synchronized section to prevent deadlocks
             Set<BaseSession>      copy = new HashSet<BaseSession>(set);
             Iterator<BaseSession> iter = copy.iterator();

             boolean interrupt = true;

             while (iter.hasNext())
             {
                BaseSession session = iter.next();
                endSession(session, interrupt, false);
                interrupt = false;
             }
          }
       }

       /**
        * Process the reply to an inbound echo request, if a session can be found
        * for the message's context.
        *
        * @param   queue
        *          The queue which received the message.
        * @param   msg
        *          Echo request message.
        */
       void processEcho(Queue queue, Message msg)
       {
          int id = msg.getKey().getContextID();
          BaseSession session = null;
          synchronized (lock)
          {
             Set<BaseSession> set = sessionsByQueue.get(queue);
             for (BaseSession s : set)
             {
                if (id == s.getContextID())
                {
                   session = s;
                   break;
                }
             }
          }

          if (session != null)
          {
             session.echoReply(msg);
          }
       }

       // Clean up all remaining sessions and tear down network environment,
       private class Stopper
       implements Runnable
       {
          public void run()
          {
             // Wait a while before cleaning up to give remote shutdown call a
             // chance to return before tearing down the transport.
             try
             {
                Thread.sleep(1000);
             }
             catch (InterruptedException exc)
             {
             }

             try
             {
                // Collect all sessions in an independent set whose iteration will
                // not be disrupted by endSession() processing.
                Set<BaseSession> copy = null;
                synchronized (lock)
                {
                   copy = new HashSet<>();
                   for (Queue key : sessionsByQueue.keySet())
                   {
                      copy.addAll(sessionsByQueue.get(key));
                   }
                }

                // End all sessions.
                boolean interrupt = true;
                Iterator<BaseSession> iter = copy.iterator();
                while (iter.hasNext())
                {
                   endSession(iter.next(), interrupt, true);
                   interrupt = false;
                }

                boolean isRoutingNode = isRouter();

                if (isRoutingNode)
                {
                   terminateInput();
                }

                Queue.terminateQueue();

                if (isRoutingNode)
                {
                   Router.terminateQueue();
                }
             }

             finally
             {
                synchronized (stopper)
                {
                   shutdownDone = true;
                }
             }
          }
       };
    }