# RouterSessionManager.java

    /*
    ** Module   : RouterSessionManager.java
    ** Abstract : SessionManager implementation for a router network node
    **
    ** Copyright (c) 2007-2025, Golden Code Development Corporation.
    **
    ** -#- -I- --Date-- --JPRM-- --------------------------Description-------------------------------
    ** 001 ECF 20071101   @35867 Created initial version. SessionManager
    **                           implementation for a router network node.
    **                           Note: reused code from other source files.
    ** 002 ECF 20071121   @35940 Fixed initialization sequence. initialize()
    **                           is now init(), which is called from the
    **                           parent's initialize() method.
    ** 003 ECF 20071127   @36049 Added certAlias parameter to connectVirtual()
    **                           variant. If alias is null, requester-side
    **                           validation of remote server's certificate is
    **                           not performed.
    ** 004 ECF 20071207   @36275 Removed try-catch in Incoming constructor.
    **                           Only IOException can be thrown, which was
    **                           already in the throws clause of the c'tor. We
    **                           were losing information about the error,
    **                           since only the error message was being stored
    **                           in the new IOException being thrown.
    ** 005 ECF 20080213   @37067 Added isLocalNode(). Reports whether a given
    **                           InetSocketAddress represents the local P2J
    **                           node.
    ** 006 GES 20081024   @40330 Insecure socket support. Refactored to allow
    **                           multiple listening loops.
    ** 007 JJC 20090208   @41405 Added getSSLSocket() method.
    ** 008 ECF 20090205   @42103 Changed to accommodate updated Queue API.
    **                           Queue.stop() method now accepts a boolean
    **                           parameter to optionally switch security
    **                           context to the session's user's context.
    ** 009 SVL 20090715   @43196 Added OOME protection into listenWorker() and
    **                           Incoming.run().
    ** 010 SVL 20090717   @43227 Server socket is restarted into listenWorker() if
    **                           it has been closed because of error.
    ** 011 GES 20090722   @43329 Mark the insecure listening thread as daemon. Exit
    **                           the listenWorker() in the case of a clean shutdown.
    **                           This was regressed by H010.
    ** 012 GES 20090722   @43337 Better thread names to enhance debugging.
    ** 013 GES 20090816   @43638 Protect createDirectSession() from executing if
    **                           the queue is stopped before the create can occur.
    **                           Added a cleanup method that clears a leak in the
    **                           incoming map in that same race condition.
    ** 014 NVS 20090816   @43645 Logging non-failure type early client exits during
    **                           authentication.
    ** 015 NVS 20090820   @43709 With recent changes to the NET package, the
    **                           connection requestor now has responsibility
    **                           to mark the queue as initialized for session.
    ** 016 CA  20090824   @43720 In Incoming.run, the logging level for
    **                           "authentication failure" message was changed from
    **                           WARNING to FINER.
    ** 017 CA  20090825   @43755 Fixed a deadlock between Queue and SessionManager.
    **                           This was discovered during remote DB connection
    **                           closing.
    ** 018 CA  20090901   @43815 Changes related to sessionsByContext rename. When
    **                           the context is set for a virtual session, it will
    **                           create and set in the context a control for this
    **                           virtual session. Fixed a leak related to the
    **                           virtual session deregistration (they were not
    **                           removed from the virtualSessionsByContext map, as
    **                           their context ID is the peer's ID, not the local
    **                           one).
    ** 019 CA  20091118   @44422 Fixed a deadlock encountered when the client
    **                           disconnects while the router builds the direct
    **                           session.
    ** 020 CA  20091202   @44472 Rolled back H019 (@44422). The real deadlock was
    **                           with RouterSessionManager.connectVirtual. All
    **                           nested locks should sync on Queue instance first
    **                           and on SessionManager.lock second.
    ** 021 CA  20100329   @44782 Added terminateVirtual implementation, which
    **                           ensures proper synchronization is done using a
    **                           distinct lock per each virtual queue, before
    **                           terminating the specified session.
    ** 022 CA  20110105          Changes to allow virtual sessions from leaf nodes
    ** 023 LMR 20101210          Changed init(), connectVirtual() and listen() to
    **                           learn the ports to open from net/server/port,
    **                           net/server/insecure_port, net/server/secure_port
    **                           in the right order, depending on
    **                           net/connection/secure and
    **                           net/connection/insecure bootstrap config entries
    **                           or command line options (net/connection/insecure
    **                           only applies to the server). Secure and insecure
    **                           environments are set up independently; if not
    **                           set, net/connection/secure defaults to false and
    **                           net/connection/insecure to true and they can't
    **                           be both set to false; a ConfigurationException
    **                           is thrown if proper port options are found to be
    **                           missing.
    ** 024 CA  20110609          Fixes a bug which deadlocked the socket listener
    **                           thread if a misconfigured client was sending the
    **                           wrong data to the socket (i.e. secure to an
    **                           insecure port and vice versa).
    ** 025 CA  20110802          Fixed a deadlock which occurs if the user closes
    **                           the terminal window right after authentication
    **                           process has started (i.e. right after ENTER was
    **                           pressed to send the username and password).
    ** 026 CS  20130204          Added java.net.Socket import to disambiguate from
    **                           com.goldencode.p2j.util.Socket class.
    ** 027 CA  20130529          Allow virtual connections to authenticate using an explicit account;
    **                           needed by appserver implementation.
    ** 028 CA  20130704          Changed "security:account:name" to the "access:subject:id", when
    **                           specifying an explicit account in connectVirtual.
    **                           Start appservers automatically, if needed.
    ** 029 OM  20131008          Added listeners to the new session created by createDirectSession().
    ** 030 CA  20140206          Added a way of notifying when the server sockets are ready (needed by
    **                           the scheduler, to know when to start launching processes).
    ** 031 EVL 20160225          Javadoc fixes to make compatible with Oracle Java 8 for Solaris 10.
    ** 032 IAS 20160331          Moved non-null assertion to a proper place
    ** 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  20190613          Fixed a deadlock related to allHosts - this must be populated in a
    **                           lock.
    ** 035 CA  20190902          The context for the requester session must be dedicated to it and not
    **                           inherited, as the session is on its own queue - and when the queue is
    **                           terminating, it will switch the activeSession, which will leak to
    **                           the inherited context's session (if setup like this).
    ** 036 IAS 20200722          Refactored to the new NetSocket API, added handshake error handling
    **     IAS 20210325          Added NIO configuration via BootstrapConfig
    **     IAS 20210827          Added allowed ciphers' filtering
    **     IAS 20220311          Correct processing of unknown host
    **     TJD 20220713          Synchronization and locking optimizations
    ** 037 GBB 20230512          Logging methods replaced by CentralLogger/ConversionStatus.
    ** 038 GBB 20230825          SecurityManager context & virtual connection method calls updated.
    ** 039 GBB 20240709          Hard-coded config names replaced by ConfigItem constants.
    ** 040 CA  20240910          'net:socket:nio' is enabled by default only for Java 8.
    ** 041 TJD 20231012          SessionManager.lock access optimizations
    ** 042 RNC 20250113          Added locking for session launching.
    ** 043 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.net.Socket;
    import java.nio.*;
    import java.nio.channels.*;
    import java.util.*;
    import java.util.concurrent.*;
    import java.util.concurrent.atomic.*;
    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.ErrorManager;
    import com.goldencode.p2j.util.logging.*;

    /**
     * Concrete implementation of a session manager for a router network node.
     * This class overrides all of the methods of <code>SessionManager</code>
     * which have behavior specific to a router node, such as security context
     * management, virtual session multiplexing and management, integration with
     * the node's router, and server-side socket connection handling.
     * <p>
     * An inner class manages new, incoming socket connections, creating a new
     * message transport queue for each.  These resources are likewise cleaned up
     * when each connection is finished.
     * <p>
     * This class is not instantiated directly.  A singleton instance is created
     * using the {@link SessionManagerFactory#createRouterNode(BootstrapConfig)}
     * factory method. This instance is later retrieved with {@link
     * SessionManager#get()}.
     */
    public final class RouterSessionManager
    extends SessionManager
    {
       /** Logger. */
       private static final CentralLogger LOG = CentralLogger.get(RouterSessionManager.class);

       /** Thread group for the connection threads. */
       private static final ThreadGroup connThreads = new ThreadGroup(
          Thread.currentThread().getThreadGroup(), "connection");

       /** Map of queues to contexts pending direct session creation. */
       private static final Map<Queue, Incoming> incomingMap =
          Collections.synchronizedMap(new HashMap<Queue, Incoming>());

       /** Server socket upon which all secure client connections arrive. */
       private ServerSocket secure = null;

       /** Server socket upon which all INSECURE client connections arrive. */
       private ServerSocket insecure = null;

       /** All host addresses on all network interfaces for the local machine. */
       private Set<InetAddress> allHosts = null;

       /** Shutdown flag. */
       private boolean shutdown = false;

       /** Signals when the network portions of the server are ready. */
       private CountDownLatch ready = new CountDownLatch(1);

       private static final AtomicInteger INPUT_COUNT = new AtomicInteger(0);

       /** Lock used to stop the launching of new sessions. If {@code true}, all incoming socket connections
        * are ignored. */
       private boolean sessionLock = false;

       /**
        * Constructor which enforces the singleton pattern.
        *
        * @throws  IllegalStateException
        *          if called more than once per JVM instance.
        * @throws  NullPointerException
        *          if <code>config</code> is <code>null</code>.
        */
       RouterSessionManager()
       {
          super();
       }

       /**
        * Expose the SSLServerSocket so the port and other parameters
        * used by the socket can be exposed to logic outside this class
        *
        * @return  <code>SSLServerSocket</code>.
        */
       public ServerSocket getSSLSocket()
       {
          return secure;
       }

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

       /**
        * 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.
        *
        * @param   address
        *          Socket address to check.
        *
        * @return  <code>true</code> if an address match is detected, else
        *          <code>false</code>.
        */
       public boolean isLocalNode(InetSocketAddress address)
       {
          InetAddress host = address.getAddress();
          if (host == null)
          {
             ErrorManager.recordOrThrowError(
                      new int[] {5193},
                      new String[] {
                         String.format(
                            "Unable to find host %s in file HOSTS or HOSTS file not found in expected location",
                            address.getHostString()
                         )
                      },
                      false, false, true
             );
          }

          int port = address.getPort();

          int securePort   = -1;
          int insecurePort = -1;

          if (secure == null)
          {
             // Not yet listening;  pull value from bootstrap config.
             securePort   = config.getInt(ConfigItem.SERVER_PORT, -1);
             insecurePort = config.getInt(ConfigItem.SERVER_INSECURE_PORT, -1);
          }
          else
          {
             // Get port from server socket(s).
             securePort = secure.getLocalPort();

             if (insecure != null)
             {
                insecurePort = insecure.getLocalPort();
             }
          }

          if (port != securePort && port != insecurePort)
          {
             return false;
          }

          // as there is a dedicated lock object, we dont synchronize on "this" but on that lock
          boolean result = false;
          synchronized (lock)
          {
             if (allHosts == null)
             {
                allHosts = new HashSet<InetAddress>();

                try
                {
                   Enumeration<NetworkInterface> e1 =
                      NetworkInterface.getNetworkInterfaces();
                   while (e1.hasMoreElements())
                   {
                      NetworkInterface ni = e1.nextElement();
                      Enumeration<InetAddress> e2 = ni.getInetAddresses();
                      while (e2.hasMoreElements())
                      {
                         InetAddress ia = e2.nextElement();
                         allHosts.add(ia);
                      }
                   }
                }
                catch (SocketException exc)
                {
                   throw new RuntimeException("Network error", exc);
                }
             }
             result = allHosts.contains(host);
          }

          return result;
       }

       /**
        * Establish a virtual (multiplexed) connection to a network node at the
        * given address.
        * <p>
        * This method is called by the requester side to establish a virtual
        * session.  The first time it must connect to a particular destination, a
        * message transport queue is established.  This requires authentication
        * between two router nodes, which is the first level of a two-level
        * authentication process.  Thereafter, requests for additional sessions
        * with that endpoint use the existing queue, and communications are
        * multiplexed by context ID.
        * <p>
        * Once a transport is available, the current user's account identifier is
        * used to authenticate at the remote node.  This user authentication
        * represents the second level of authentication.  On the remote side, if
        * authentication is successful (the account must be registered under the
        * same ID on the remote side, and have sufficient rights to authenticate),
        * the other "half" of the virtual session is established on the remote
        * side.  A unique context ID is assigned to that session and is returned
        * to the requester.  This context ID is used to establish a virtual
        * session ({@link RequesterSession}) on this side.  This ID is later used
        * in outbound messages to multiplex communications over the shared queue.
        * <p>
        * This method creates an in-memory <code>BootstrapConfig</code> object to
        * enable the first level of authentication, which is certificate-based.
        * Configuration entries are populated as follows (note that some values
        * are retrieved from the <code>BootstrapConfig</code> instance set by
        * {@link SessionManager#initialize(BootstrapConfig)};  this version of the
        * configuration is referred to below as the <i>default config</i>):
        * <ul>
        *   <li>server mode is set to false;  even though the requester is a
        *       router node, it is treated as a client from the security manager's
        *       perspective when requesting authentication
        *   <li>net/server/host - taken from <code>address</code> parameter
        *   <li>net/server/port - taken from <code>address</code> parameter
        *   <li>security/certificate/validate - <code>true</code> only if
        *       <code>certAlias</code> is non-<code>null</code>, otherwise is left
        *       unset, which defaults to <code>false</code> with the security
        *       manager;  if <code>true</code>, forces certificate-based
        *       validation of the providing server
        *   <li>security/truststore/alias - taken from <code>certAlias</code>
        *       parameter, if that parameter is non-<code>null</code>, otherwise
        *       is left unset;  this is necessary to allow the security manager to
        *       check the correct truststore for the providing server, if
        *       certification-based validation of the providing server is desired
        * </ul>
        *
        * @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.  May be
        *          <code>null</code>, which indicates requesting server does not
        *          validate providing server's certificate.
        * @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  Exception
        *          if there are any errors reading configuration, creating a
        *          queue, or authenticating the user session remotely.
        */
       public Session connectVirtual(InetSocketAddress address,
                                     String            certAlias,
                                     SessionListener   notify,
                                     String            account)
       throws Exception
       {
          String host = address.getHostName();
          int port = address.getPort();

          BootstrapConfig config = new BootstrapConfig();
          config.setServer(false);

          // Network category.
          config.setConfigItem(ConfigItem.SERVER_HOST, host);
          config.setConfigItem(ConfigItem.SERVER_SECURE_PORT, String.valueOf(port));

          // virtual connections are always made over secure sockets
          config.setConfigItem(ConfigItem.SECURE, String.valueOf(true));

          // Security category.
          if (certAlias != null)
          {
             // Enable requester-side validation of remote server's certificate.
             config.setConfigItem("security", "certificate", "validate", "true");
             config.setConfigItem("security", "truststore", "alias", certAlias);
          }

          if (account != null)
          {
             config.setConfigItem("access", "subject", "id", account);
          }

          return connectVirtual(config, notify);
       }

       /**
        * Establish a virtual (multiplexed) connection to a network node at the
        * given address.
        * <p>
        * This method is called by the requester side to establish a virtual
        * session.  The first time it must connect to a particular destination, a
        * message transport queue is established.  This requires authentication
        * between two router nodes, which is the first level of a two-level
        * authentication process.  Thereafter, requests for additional sessions
        * with that endpoint use the existing queue, and communications are
        * multiplexed by context ID.
        * <p>
        * Once a transport is available, the current user's account identifier is
        * used to authenticate at the remote node.  This user authentication
        * represents the second level of authentication.  On the remote side, if
        * authentication is successful (the account must be registered under the
        * same ID on the remote side, and have sufficient rights to authenticate),
        * the other "half" of the virtual session is established on the remote
        * side.  A unique context ID is assigned to that session and is returned
        * to the requester.  This context ID is used to establish a virtual
        * session ({@link RequesterSession}) on this side.  This ID is later used
        * in outbound messages to multiplex communications over the shared queue.
        * <p>
        * This method is synchronized so, if multiple clients are calling this
        * method, only one of them will create the connection.
        *
        * @param   config
        *          Bootstrap configuration information which contains at least
        *          enough information to make a virtual connection from this node
        *          to another.  See {@link #connectVirtual(BootstrapConfig, SessionListener)}
        *          for details.
        * @param   notify
        *          Session listener to be notified of session events.
        *
        * @return  Session established after the connection has been made.
        *
        * @throws  Exception
        *          if there are any errors reading configuration, creating a
        *          queue, or authenticating the user session remotely.
        */
       public Session connectVirtual(BootstrapConfig config,
                                     SessionListener notify)
       throws Exception
       {
          // removed synchronized keyword as there is a lock object to synchronize on
          // net/server/secure_port is only read if net/connection/secure==true
          if (!config.getBoolean(ConfigItem.SECURE, false))
          {
             throw new ConfigurationException("net/connection/secure is false");
          }

          int    port = config.getInt(ConfigItem.SERVER_SECURE_PORT, -1);

          String host = config.getString(ConfigItem.SERVER_HOST, null);
          InetSocketAddress remoteAddr = new InetSocketAddress(host, port);
          Queue queue = null;
          boolean createdQueue = false;

          synchronized (lock)
          {
             // during queue creation there is no lock on SM.lock, so its safe to
             // sync on SM.lock here
             queue = topology.get(remoteAddr);

             if (queue == null)
             {
                queue = this.createQueue(remoteAddr, config);
                createdQueue = true;

                virtualLocksByQueue.put(queue, new Object());
             }
          }

          if (createdQueue)
          {
             // no synchronization on SM.lock is needed for queue start
             int nodeAddr = this.config.getInt("net",
                   "server",
                   "nodeAddr",
                   0x00010000);
             queue.setNodeAddress(nodeAddr);

             // Begin protocol operations.
             int timeout = config.getInt("net", "server", "timeout", 600000);
             queue.start(true, timeout, false, false);
          }

          BaseSession session = null;

          try
          {
             // no synchronization on SM.lock is needed for authentication
             // first check if an explicit account is not set
             String identity = config.getConfigItem("access", "subject", "id");
             if (identity == null || identity.equals(""))
             {
                Object context = secMgr.virtualSm.getEffectiveContext();
                // use the account for the current context, if not explicitly set
                identity = secMgr.virtualSm.getIdentity(context);
             }
             assert (!"".equals(identity));
             int contextID = nextContextID();
             int peerID = callAuthenticateRemote(identity, contextID, queue);

             // registerSession sets a lock on SM.lock, so a lock on queue must be
             // acquired first
             synchronized (queue)
             {
                Object virtualContext = secMgr.virtualSm.createSecurityContext(identity);
                if (virtualContext == null)
                {
                   throw new IllegalAccessException("Could not create virtual session.");
                }
                session = new RequesterSession(queue, virtualContext, peerID, contextID);
                queue.sessionInitialized();
                if (notify != null)
                {
                   session.addSessionListener(notify);
                }

                // Update session manager database.
                registerSession(contextID, remoteAddr, session);
             }

             // At this moment the session is almost ready for use, let the listeners the
             // initialization has ended
             session.sendInitializationEvent();
          }
          catch (IllegalAccessException exc)
          {
             if (createdQueue)
             {
                // remote authentication was denied;  if we created the queue
                // and no one else is using it, tear it down

                Set<BaseSession> sessions = null;
                synchronized (lock)
                {
                   sessions = sessionsByQueue.get(queue);
                }
                // moved out synchronized section to prevent deadlocks
                if (sessions == null)
                {
                   queue.stop(exc, false);
                }
             }

             exc.fillInStackTrace();

             throw exc;
          }

          return session;
       }

       /**
        * 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 calling thread was interrupted
        */
       public void waitUntilReady()
       throws InterruptedException
       {
          ready.await();
       }

       /**
        * Listen for incoming socket connections, authenticate each new
        * connection, and establish a new queue for each authenticated connection.
        * <p>
        * Depending on which modes (secure/insecure) are enabled, this method
        * will start one or two socket listening loops. The insecure socket
        * can be used to obtain better performance but must only be used in cases
        * where the network transport is known to be secure. This is notoriously
        * difficult to ensure and so such usage should be avoided if at all
        * possible.
        * <p>
        * If both insecure and secure sockets are used, a second thread will be
        * launched to implement the secure listening loop.
        *
        * @param   notify
        *          Session listener to be notified of session events.
        *
        * @throws  Exception
        *          if there is any configuration error, or error creating a server
        *          socket.
        */
       public void listen(final SessionListener notify)
       throws Exception
       {
          // see which ports have to be opened: secure and/or non-secure
          // (init() checks to make sure that (isSecure || isInsecure)==true)
          boolean isSecure = this.config.getBoolean(ConfigItem.SECURE, false);
          boolean isInsecure = this.config.getBoolean("net", "connection", "insecure", true);

          //optionally init the non-secure environment
          if (isInsecure)
          {
             int insecurePort = config.getInt(ConfigItem.SERVER_PORT, -1);
             if (insecurePort == -1)
             {
                insecurePort = config.getInt(ConfigItem.SERVER_INSECURE_PORT, -1);
             }
             createServerSocket(false, insecurePort);
             if (LOG.isLoggable(Level.WARNING))
             {
                LOG.logp(Level.WARNING,
                         "SessionManager.listen()",
                         "",
                         "INSECURE sockets in use!");
             }
          }

         // optionally init the secure environment
         if (isSecure)
         {
            int securePort = config.getInt(ConfigItem.SERVER_SECURE_PORT, -1);
            if (securePort != -1)
            {
               createServerSocket(true, securePort);

               // start the secondary listening thread if needed
               if (isInsecure)
               {
                  Thread t = new Thread(connThreads, new Runnable()
                  {
                     public void run()
                     {
                        listenWorker(notify, secure, Mode.SSL);
                     }
                  });

                  t.setDaemon(true);
                  t.setName("Secure Socket Listener");
                  t.start();
               }
            }
            // if the insecure mode is not enabled the current thread listens on
            // the secure socket
          }

          // log this before the current thread starts its listening loop
          if (LOG.isLoggable(Level.INFO))
          {
             LOG.logp(Level.INFO,
                      "SessionManager.listen()",
                      "",
                      "Server ready");
          }

          ready.countDown();

          // start listening loop in the current thread (secure or non-secure):
          if (isInsecure)
          {
             // if enabled, the secure listening loop has been already started in
             // a separate thread
             listenWorker(notify, insecure, Mode.INSECURE);
          }
          else
          {
             // secure mode must've been enabled (we tested for this in init())
             listenWorker(notify, secure, Mode.SSL);
          }
       }

       /**
        * 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 (super.toString() + " PENDING: " + incomingMap);
       }

       /**
        * Setter for the {@link #sessionLock} variable.
        *
        * @param   lockStatus
        *          The new status for the lock. {@code true} if the session launching will be stopped,
        *          {@code false} otherwise
        *
        * @return  The new value of the {@link #sessionLock} variable.
        */
       @Override
       public synchronized boolean setSessionLock(boolean lockStatus)
       {
          sessionLock = lockStatus;
          return sessionLock;
       }

       /**
        * Getter for the {@link #sessionLock} variable.
        *
        * @return  The value of the {@link #sessionLock} variable.
        */
       @Override
       public synchronized boolean getSessionLock()
       {
          return sessionLock;
       }

       /**
        * 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.
        *
        * @param   identity
        *          Account identifier for the requesting party.
        * @param   peerID
        *          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  IllegalAccessException
        *          if authentication is not successful.
        */
       public int authenticateRemote(String identity, int peerID, Queue queue)
       throws IllegalAccessException
       {
          incomingMap.remove(queue);

          Object context = null;

          try
          {
             context = secMgr.authenticateRemote(identity);
          }
          catch (RestrictedUseException exc)
          {
             // safe calling location;  should not be thrown
          }

          if (context == null)
          {
             throw new IllegalAccessException(
               "Remote access for " + identity + " is not allowed");
          }

          int contextID = nextContextID();

          VirtualSession session = new VirtualSession(queue,
                                                      context,
                                                      peerID,
                                                      contextID);
          registerSession(contextID, null, session);

          // At this moment the session is almost ready for use, let the listeners the
          // initialization has ended
          session.sendInitializationEvent();

          return contextID;
       }

       /**
        * 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.
        *
        * @param   contextID
        *          Value which uniquely identifies the security context of the
        *          requesting party.
        */
       public void terminateRemote(int contextID)
       {
          Queue queue = null;

          // determine queue first, so locking by queue is possible
          synchronized (lock)
          {
             BaseSession session = virtualSessionsByContext.get(contextID);
             assert (session != null);
             queue = session.getQueue();
          }

          synchronized (queue)
          {
             BaseSession session = null;
             synchronized (lock)
             {
                session = virtualSessionsByContext.get(contextID);
                assert (session != null);
             }
             // moved out synchronized section to prevent deadlocks
             // endSession can be called out of SM.lock
             endSession(session, true, true);
          }
       }

       /**
        * Perform any special initialization which is not handled by the default
        * (common) logic.  This includes checking the bootstrap configuration for
        * valid values, initializing the dispatcher, and initializing the router.
        *
        * @throws  IllegalStateException
        *          if the router node is already listening for connections.
        * @throws  ConfigurationException
        *          if any configuration error occurs during initialization.
        */
       protected void init()
       throws ConfigurationException
       {
          if (secure != null ||  insecure != null)
          {
             throw new IllegalStateException("The server is already running.");
          }

          boolean isSecure   = this.config.getBoolean(ConfigItem.SECURE, false);
          boolean isInsecure = this.config.getBoolean("net", "connection", "insecure", true);
          if (!isSecure && !isInsecure)
          {
             throw new ConfigurationException("net/connection/secure and " +
                   "net/connection/insecure are both false");
          }
          incomingMap.clear();

          String serverId = config.getString("security", "server", "id", null);
          if (serverId == null || serverId.length() == 0)
          {
             throw new ConfigurationException("Missing server ID.");
          }

          int nodeAddr = config.getInt("net", "server", "nodeAddr", 0x00010000);
          if (nodeAddr == 0 || (nodeAddr & AddressServer.NODE_MASK) != 0)
          {
             throw new ConfigurationException("Missing or invalid node address.");
          }

          int port;
          if (isInsecure)
          {
             port = config.getInt(ConfigItem.SERVER_PORT, -1);
             if (port <= 0)
             {
                port = config.getInt(ConfigItem.SERVER_INSECURE_PORT, -1);
             }
             if (port <= 0)
             {
                throw new ConfigurationException("Missing or invalid insecure port number.");
             }
          }
          if (isSecure)
          {
             port = config.getInt(ConfigItem.SERVER_SECURE_PORT, -1);
             if (port <= 0)
             {
                throw new ConfigurationException("Missing or invalid secure port number.");
             }
          }

          // initialize the protocol's dispatcher
          int threads = config.getInt("net", "dispatcher", "threads", 2);
          Dispatcher.initialize(threads);

          // init router
          int rthreads = config.getInt("net", "router", "threads", 1);
          Router.initRouter(nodeAddr, rthreads);
       }

       /**
        * Create a direct session for the given queue and register it with the
        * session manager.  This enables the deferred creation of a direct
        * session on the provider side of the connection, during the requester's
        * queue's handshaking with the router.  This deferral is necessary
        * because there is not enough information at the time the connection is
        * established and the queue is created, to determine what type of session
        * is needed.  The security context, which is temporarily stored during
        * connection time (when the requester is authenticated), is now associated
        * with the new session and is removed from its temporary storage.
        *
        * @param    queue
        *           Underlying message transport for the new session.
        *
        * @return   <code>true</code> if the session initialized successfully.
        */
       protected boolean createDirectSession(Queue queue)
       {
          synchronized (queue)
          {
             // removed internal synchronized section to prevent deadlocks
             // if the session was aborted before we could initialize, there is
             // nothing to do
             if (!queue.isRunning())
                return false;

             Incoming incoming = incomingMap.remove(queue);
             assert (incoming != null);

             DirectSession session = new DirectSession(queue, incoming.context);
             if (incoming.notify != null)
             {
                session.addSessionListener(incoming.notify);
             }
             registerSession(0, null, session);

             // add saved callbacks for initialization and termination of session
             for (SessionListener sl : incoming.socket.getSessionListeners())
             {
                session.addSessionListener(sl);
             }
          }

          return true;
       }

       /**
        * Close the server's socket if it was ever initialized.
        */
       protected void terminateInput()
       {
          try
          {
             synchronized (lock)
             {
                shutdown = true;
             }

             if (secure != null)
             {
                secure.close();
             }

             if (insecure != null)
             {
                insecure.close();
             }

             // wait a while after terminating
             Thread.sleep(1000);
          }
          catch (InterruptedException exc)
          {
          }

          catch (IOException exc)
          {
             // log the fact that some form of failure has occurred
             if (LOG.isLoggable(Level.WARNING))
             {
                LOG.logp(Level.WARNING,
                         "RouterSessionManager.terminateInput()",
                         "",
                         "socket close failure",
                         exc);
             }
          }
       }

       /**
        * 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>
        * The parent's implementation is overridden here to permit additional
        * cleanup necessary for this implementation.  However, this implementation
        * first invokes the parent's.
        *
        * @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)
       {
          boolean queueInUse = super.deregisterSession(session);
          Queue queue = session.getQueue();

          assert (queue != null);

          synchronized (queue)
          {
             synchronized (lock)
             {
                int contextID = session.getContextID();
                // if this is a virtual session, we must get the ID from the local
                // peer ID
                if (session instanceof VirtualSession)
                {
                   contextID = ((VirtualSession) session).getLocalContextID();

                   virtualSessionsByContext.remove(contextID);
                   virtualControlByContext.remove(contextID);
                }
             }
          }

          if (!queueInUse)
          {
             incomingMap.remove(queue);
          }

          return queueInUse;
       }

       /**
        * Set the security context for the current thread, using the session
        * identified by the given routing key or transport queue.  If the routing
        * key has a security context ID of 0, the context of the default session
        * for the given queue is used.  If the routing key's context ID is
        * non-zero, the appropriate session is retrieved for the ID and its
        * context is used.
        *
        * @param   key
        *          Routing key for an incoming message.
        * @param   queue
        *          Message transport queue.
        */
       protected void setContext(RoutingKey key, Queue queue)
       {

          int contextID = key.getContextID();
          if (contextID == 0)
          {
             // No demultiplexing required;  there is only one session for the
             // queue.
             setContext(queue);
          }
          else
          {
             // Demultiplex the session by context ID from among the sessions
             // sharing the queue.
             BaseSession session = null;

             synchronized (lock)
             {
                session = virtualSessionsByContext.get(contextID);
             }

             setContext(session);

             synchronized (lock)
             {
                // get the control for this context
                Control ct = Control.locate(contextID);

                if (!ct.isReady())
                {
                   ct.initControl(session);
                }

                Control.setContext(ct);
             }
          }
       }

       /**
        * 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.
        *
        * @param   queue
        *          Queue which uniquely identifies the session whose security
        *          context is to be used.
        *
        * @throws  IllegalArgumentException
        *          if there is not exactly one session for the given queue.
        */
       protected void setContext(Queue queue)
       {
          BaseSession session = null;

          synchronized (queue)
          {
             synchronized (lock)
             {
                Set<BaseSession> set = sessionsByQueue.get(queue);

                // this is a special case of setting the context after the queue is
                // stopped: if the requester side terminated abnormally and it had a
                // chance to establish more than one virtual session, the remote side
                //  will "see" more than one session for this queue; thus, in this
                // case, we will use the associated DirectSession as the source of
                // the context, and fail if there is more than one DirectSession for
                // this queue
                if (!queue.isRunning() && set != null)
                {
                   Iterator<BaseSession> iter = set.iterator();
                   boolean found = false;
                   while (iter.hasNext())
                   {
                      BaseSession sess = iter.next();
                      if (sess instanceof DirectSession)
                      {
                         if (!found)
                         {
                            found = true;
                            session = sess;
                         }
                         else
                         {
                            throw new IllegalArgumentException(
                               "No default session found for queue");
                         }
                      }
                   }
                }

                if (session == null)
                {
                   if (set == null || set.size() != 1)
                   {
                      throw new IllegalArgumentException(
                         "No default session found for queue");
                   }

                   session = set.iterator().next();
                }
             }
          }

          setContext(session);
       }

       /**
        * 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.
        */
       @Override
       protected void setContext(BaseSession session)
       {
          // Set the session's current context.
          session.setContext();

          // Remember the current session.
          activeSession.set(session);
       }

       /**
        * Restore the current thread's security context to whatever it was before
        * it was last explicitly set.
        */
       protected void restoreContext()
       {
          try
          {
             secMgr.contextSm.popAndRestoreSecurityContext();
          }
          catch (RestrictedUseException exc)
          {
             throw new RuntimeException("Invalid security context switch");
          }
       }

       /**
        * Initialize security context switching for the calling thread.
        */
       protected void setInitialContext()
       {
          try
          {
             secMgr.contextSm.setInitialSecurityContext();
          }
          catch (RestrictedUseException exc)
          {
             throw new RuntimeException("Invalid security context switch");
          }
       }

       /**
        * Release the initial security context assigned to the calling thread.
        */
       protected void dropInitialContext()
       {
          secMgr.contextSm.dropInitialSecurityContext();
       }

       /**
        * Cleanup session state before initialization.
        *
        * @param    queue
        *           The queue that represents the partially initialized session
        *           to be cleaned up.
        *
        * @return   Security manager context to that still requires cleanup.
        */
       Object abortSessionInit(Queue queue)
       {
          Incoming incoming = incomingMap.remove(queue);
          return incoming == null ? null : incoming.context;
       }

       /**
        * Core listening loop for incoming socket connections.
        *
        * @param   notify
        *          Session listener to be notified of session events.
        * @param   input
        *          The socket on which to listen.
        * @param   mode
        *          The connection mode.
        */
       private void listenWorker(SessionListener notify, ServerSocket input, Mode mode)
       {
          String transientName = String.format("Incoming %s Connector", mode);

          String logtxt = String.format("SessionManager.listenWorker(%s)", mode);

          // main socket listening loop
          while (true)
          {
             Socket sock = null;

             SocketChannel channel = null;
             ServerSocketChannel serverSocketChannel = input.getChannel();
             try
             {
                Incoming incoming;
                if (serverSocketChannel != null)
                {
                   channel = serverSocketChannel.accept();
                   channel.setOption(StandardSocketOptions.TCP_NODELAY , true);
                   incoming = new Incoming(channel, mode, notify);
                }
                else
                {
                   sock = input.accept();
                   sock.setTcpNoDelay(true);
                   incoming = new Incoming(sock, mode, notify);
                }

                // we need to authenticate this incoming connection before
                // starting the queue, since this is a potentially long-running
                // task we create a thread that does this and then that thread
                // exits (this way this listening loop is not blocked)
                Thread t = new Thread(connThreads, incoming);

                // make sure this temporary thread doesn't stop the server from
                // exiting
                t.setDaemon(true);
                t.setName(transientName + "#" + INPUT_COUNT.incrementAndGet());

                t.start();
             }

             catch (Throwable thr)
             {
                // log the fact that some form of failure has occurred

                try
                {
                   boolean clean = false;

                   synchronized (lock)
                   {
                      clean = shutdown;
                   }

                   if (clean)
                   {
                      if (LOG.isLoggable(Level.INFO))
                      {
                         LOG.logp(Level.INFO,
                                  logtxt,
                                  "",
                                  "Server shutting down.");
                      }

                      // exit the listener
                      return;
                   }
                   else
                   {
                      if (LOG.isLoggable(Level.WARNING))
                      {
                         LOG.logp(Level.WARNING,
                                  logtxt,
                                  "",
                                  "Failure in listening loop.",
                                  thr);
                      }
                   }

                   // release the connection with the recently connected client
                   if (sock != null)
                   {
                      try
                      {
                         sock.close();
                      }
                      catch (IOException e)
                      {
                         // Ignore.
                      }
                   }

                   if (thr instanceof OutOfMemoryError)
                   {
                      try
                      {
                         // Some delay to let OOME dissapear.
                         Thread.sleep(5000);
                      }
                      catch (InterruptedException e)
                      {
                         // Ignore.
                      }
                   }

                   // if the server socket is closed then we should restart it
                   if (input.isClosed())
                   {
                      try
                      {
                         boolean isSecure = input instanceof SSLServerSocket;
                         input = createServerSocket(isSecure,
                                                    input.getLocalPort());
                      }
                      catch (Exception e)
                      {
                         if (LOG.isLoggable(Level.SEVERE))
                         {
                            LOG.logp(Level.SEVERE,
                                     logtxt,
                                     "",
                                     "error while recreating server socket",
                                     e);
                         }
                         break;
                      }
                   }
                }
                catch (OutOfMemoryError e)
                {
                   // Ignore.
                }
             }
          }

          if (LOG.isLoggable(Level.INFO))
          {
             LOG.logp(Level.INFO,
                      logtxt,
                      "",
                      "Server complete");
          }
       }

       /**
        * Create required server socket and assign it to the corresponding
        * field-level variable: {@link #secure} or {@link #insecure}.
        *
        * @param  isSecure
        *         Determines whether the created socket should be secure.
        * @param  port
        *         The port which the socket should listen to.
        *
        * @return created server socket.
        *
        * @throws Exception
        *         if an error has occurred while creating server socket.
        */
       private ServerSocket createServerSocket(boolean isSecure, int port)
       throws Exception
       {
          InetSocketAddress endpoint = new InetSocketAddress(port);
          if (config.getBoolean(ConfigItem.SERVER_SOCKET_NIO, false))
          {
             ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
             serverSocketChannel.bind(endpoint);
             ServerSocket socket = serverSocketChannel.socket();
             if (isSecure)
             {
                secure = socket;
             }
             else
             {
                insecure = socket;
             }
             return socket;
          }
          if (isSecure)
          {
             SSLContext ctx = secMgr.contextSm.getSecureSocketContext();
             SSLServerSocketFactory factory = ctx.getServerSocketFactory();
             secure = factory.createServerSocket(port);
             ((SSLServerSocket) secure).setWantClientAuth(true);
          }
          else
          {
             insecure = new ServerSocket();
             insecure.bind(endpoint);
          }

          return isSecure ? secure : insecure;
       }

       /**
        * Connection mode
        */
       private static enum Mode
       {
          INSECURE,
          SSL
       }
       /**
        * This class performs all long running tasks with an incoming connection,
        * such as authentication and queue initialization.
        */
       private class Incoming
       implements Runnable
       {
          /** Listener for session events. */
          private final SessionListener notify;

          /** Client connection socket. */
          private NetSocket socket = null;

          /** Incoming connection socket. */
          private final Socket inputSocket;

          /** The input channel. */
          private final SocketChannel inputChannel;

          /** Security context */
          private Object context = null;

          /** The connection mode. */
          private Mode mode;

          /**
           * Constructs an instance.
           *
           * @param   socket
           *          Incoming session's transport.
           * @param   mode
           *          The connection mode.
           * @param   notify
           *          Listener to receive session event notifications.  May be
           *          <code>null</code>.
           */
          Incoming(Socket socket, Mode mode, SessionListener notify)
          {
             this.inputChannel = null;
             this.inputSocket = socket;
             this.notify = notify;
             this.mode = mode;
          }

          /**
           * Constructs an instance.
           *
           * @param   channel
           *          Incoming session's channel.
           * @param   mode
           *          The connection mode.
           * @param   notify
           *          Listener to receive session event notifications.  May be
           *          <code>null</code>.
           */
          Incoming(SocketChannel channel, Mode mode, SessionListener notify)
          {
             this.inputChannel = channel;
             this.inputSocket = channel.socket();
             this.notify = notify;
             this.mode = mode;
          }

          /**
           * Main processing thread. It does the following:
           * <ul>
           *    <li>initializes security context</li>
           *    <li>authenticates client</li>
           *    <li>starts queue</li>
           * </ul>
           */
          public void run()
          {
             SessionManager sessMgr = SessionManager.get();
             Queue queue = null;
             boolean inContext = false;

             LOG.log(Level.FINE, "New incoming connection: " + Thread.currentThread().getName());
             try
             {
                if (inputChannel == null)
                {
                   this.socket = NetSocket.FACTORY.create(inputSocket);
                }
                else if (mode == Mode.INSECURE)
                {
                   this.socket = NetSocket.FACTORY.create(inputChannel);
                }
                else
                {
                      SSLContext ctx = secMgr.contextSm.getSecureSocketContext();
                      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(false);
                      engine.setNeedClientAuth(false);
                      engine.setWantClientAuth(true);
                      this.socket = NetSocket.FACTORY.create(engine, inputChannel);
                }

                if (getSessionLock())
                {
                   try
                   {
                      if (socket != null && !socket.isClosed())
                      {
                         // first read the AUTHTYPE sent by the client
                         socket.readInt();

                         // then send the locking status
                         socket.writeInt(SecurityConstants.AUTH_MODE_NONE);
                         socket.writeInt(SecurityConstants.AUTH_ACTION_LOCKED);
                         socket.flush();
                         socket.close();
                      }
                   }
                   catch (IOException e)
                   {
                      // The socket was most probably closed.
                   }
                   return;
                }

                // obtain an initial security context
                sessMgr.setInitialContext();
                inContext = true;

                // authenticate client
                context = sessMgr.secMgr.authenticateLocal(socket);

                if (context != null)
                {
                   if (context instanceof String)
                   {
                      // log the fact that some form of client exit has occurred
                      if (LOG.isLoggable(Level.FINE))
                      {
                         LOG.logp(Level.WARNING,
                                  "Incoming.run()",
                                  "",
                                  (String) context);
                      }
                   }
                   else
                   {
                      queue = new Queue(socket);
                      incomingMap.put(queue, this);
                      queue.start(false, 0, false, false);
                   }
                }
                else
                {
                   // log the fact that some form of failure has occurred
                   if (LOG.isLoggable(Level.FINER))
                   {
                      LOG.logp(Level.FINER,
                               "Incoming.run()",
                               "",
                               "authentication failure");
                   }
                }
             }
             catch (Throwable thr)
             {
                // log the fact that some form of failure has occurred

                try
                {
                   if (LOG.isLoggable(Level.WARNING))
                   {
                      if (thr instanceof SSLHandshakeException)
                      {
                         LOG.log(Level.WARNING, thr.getMessage());
                      }
                      else
                      {
                         LOG.logp(Level.WARNING,
                                  "Incoming.run()",
                                  "",
                                  "initialization failure",
                                  thr);
                         }
                   }

                   if (queue != null)
                   {
                      incomingMap.remove(queue);
                      queue.stop(thr instanceof Exception ? (Exception) thr :
                                                            new Exception(thr),
                                 false);
                   }

                   if (context != null && !socket.isClosed())
                   {
                      socket.close();
                   }
                }
                catch (OutOfMemoryError e)
                {
                   // ignore
                }
             }
             finally
             {
                if (inContext)
                {
                   sessMgr.dropInitialContext();
                }
                LOG.log(Level.FINE, "Incoming connection OK: " + Thread.currentThread().getName());
             }
          }
       }
    }