# LeafSessionManager.java

    /*
    ** Module   : LeafSessionManager.java
    ** Abstract : SessionManager implementation for a leaf network node
    **
    ** Copyright (c) 2004-2024, Golden Code Development Corporation.
    **
    ** -#- -I- --Date-- --JPRM-- --------------------------- Description ---------------------------
    ** 001 ECF 20071101  @35866  Created initial version. SessionManager
    **                           implementation for a leaf network node.
    ** 002 ECF 20071121  @35939  Fixed initialization sequence. initialize()
    **                           is now init(), which is called from the
    **                           parent's initialize() method.
    ** 003 NVS 20090820  @43708  With the recent changes to the NET package, the
    **                           connection requestor now has responsibility
    **                           to mark the queue as initialized for the session.
    ** 004 CA  20090901  @43812  In connectDirect, if the received interrupt
    **                           handler is not null, will be saved, as now the
    **                           InterruptHandler implementation is saved in the
    **                           SessionManager.
    ** 005 CA  20101217          Allow multiple direct sessions on leaf nodes.
    ** 006 CA  20110105          Allow virtual session connections from leaf nodes
    **                           Once a virtual session is established, all
    **                           subsequent virtual sessions will use the same
    **                           credentials as the first authenticated one.
    **                           Once a direct or virtual session is established,
    **                           all subsequent sessions need to be of the same
    **                           type: direct or virtual.
    ** 007 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.
    ** 008 LMR 20101209          Changed init() and connectDirect() to have the
    **                           port to connect to taken from net/server/port,
    **                           net/server/insecure_port, net/server/secure_port
    **                           in the right order, depending on
    **                           net/connection/secure (if this is not present
    **                           it defaults to false).
    ** 009 OM  20131025          Sending initialization event on newly created session.
    ** 010 CA  20220515          The active session is now set for leaf sessions, too - this is required for a
    **                           change in RemoteObject.obtainInstance, where first a check is done for an existing
    **                           local proxy, and after that a network instance is obtained (a requirement for the
    **                           server-side OS resources support, like memptr).
    **     TJD 20220713          Synchronization and locking optimizations
    ** 011 GBB 20240709          Hard-coded config names replaced by ConfigItem constants.
    ** 012 TJD 20231012          SessionManager.lock access optimizations
    */
    /*
    ** 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.net.*;
    import java.util.*;
    import com.goldencode.p2j.cfg.*;
    import com.goldencode.p2j.util.*;

    /**
     * Concrete implementation of a session manager for a leaf network node.
     * This class overrides all of the methods of <code>SessionManager</code>
     * which have behavior specific to a leaf node.  Essentially, this is just
     * differences in initialization and the ability to request direct session
     * establishment with a remote network node.
     * <p>
     * This class is not instantiated directly.  A singleton instance is created
     * using the {@link SessionManagerFactory#createLeafNode(BootstrapConfig)}
     * factory method. This instance is later retrieved with {@link
     * SessionManager#get()}.
     * <p>
     * It is possible to create multiple direct sessions or multiple virtual
     * sessions to a router node. But, current implementation does not allow
     * establishing at the same time direct and virtual sessions to the same
     * router. Once all sessions to that router are terminated, the mode is reset
     * and the next session can be of either type: direct or virtual.
     */
    final class LeafSessionManager
    extends SessionManager
    {
       /**
        * Identifier indicating no session was established to that specific remote
        * address.
        */
       private static final int NOT_ESTABLISHED = 0;

       /**
        * Identifier indicating that a direct session was establies to that
        * specific remote address and all subsequent sessions must be of the same
        * type (i.e. direct).
        */
       private static final int DIRECT_MODE = 1;

       /**
        * Identifier indicating that a virtual session was establies to that
        * specific remote address and all subsequent sessions must be of the same
        * type (i.e. virtual).
        */
       private static final int VIRTUAL_MODE = 2;

       /**
        * Map containg the type of sessions which are allowed to be created for
        * the specific address.
        */
       private final Map<InetSocketAddress, Integer> connectionModes =
          new HashMap<InetSocketAddress, Integer>();

       /**
        * 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>.
        */
       LeafSessionManager()
       {
          super();
       }

       /**
        * 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 true;
       }

       /**
        * Establish a direct connection to a network node using the bootstrap
        * configuration stored during construction.
        *
        * @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  Exception
        *          if there are any errors reading configuration or creating a
        *          queue.
        */
       public Session connectDirect(BootstrapConfig config,
                                    InterruptHandler interruptHandler,
                                    SessionListener notify)
       throws Exception
       {
          if (config == null)
          {
             config = this.config;
          }

          int port = -1;
          boolean isSecure = this.config.getBoolean(ConfigItem.SECURE, false);
          if (isSecure)
          {
             port = config.getInt(ConfigItem.SERVER_SECURE_PORT, -1);
          }
          else
          {
             port = config.getInt(ConfigItem.SERVER_PORT, -1);
             if (port == -1)
             {
                port = config.getInt(ConfigItem.SERVER_INSECURE_PORT, -1);
             }
          }

          String host = config.getString(ConfigItem.SERVER_HOST, null);
          InetSocketAddress remoteAddr = new InetSocketAddress(host, port);

          synchronized (lock)
          {
             int connectionMode = connectionModes.containsKey(remoteAddr)
                                     ? connectionModes.get(remoteAddr)
                                     : NOT_ESTABLISHED;
             if (connectionMode == VIRTUAL_MODE)
             {
                throw new IllegalStateException(
                   "Only direct sessions can be established to " + remoteAddr + " at this time");
             }
             else
             {
                connectionModes.put(remoteAddr, DIRECT_MODE);
             }
          }

          if (config.isServer())
          {
             synchronized (lock)
             {
                if (topology.containsKey(remoteAddr))
                {
                   throw new IllegalStateException(
                      "Cannot create second direct connection to " + remoteAddr);
                }
             }
          }

          // create queue and session
          Queue queue = createQueue(remoteAddr, config);
          DirectSession session = new DirectSession(queue, null);
          if (notify != null)
          {
             session.addSessionListener(notify);
          }

          // initialize interrupt handling
          if (interruptHandler != null)
          {
             setInterruptHandler(interruptHandler);

             Control.init(session);
          }

          // begin protocol operations
          int timeout = config.getInt("net", "server", "timeout", 600000);
          boolean conversation = config.getBoolean("net", "queue", "conversation", false);
          boolean startThread =  config.getBoolean("net", "queue", "start_thread", false);
          queue.start(true, timeout, conversation, startThread);
          queue.sessionInitialized();

          // update session manager database
          registerSession(0, remoteAddr, session);

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

          activeSession.set(session);

          return session;
       }

       /**
        * 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();

          synchronized (queue)
          {
             synchronized (lock)
             {
                int contextID = session.getContextID();
                if (session instanceof VirtualSession)
                {
                   virtualControlByContext.remove(contextID);

                   // if this is a virtual session, we must get the ID from the
                   // local peer ID
                   contextID = ((VirtualSession) session).getLocalContextID();
                   virtualSessionsByContext.remove(contextID);
                }

                InetSocketAddress remoteAddr = queue.getRemoteInetAddress();
                if (!addressUseCount.containsKey(remoteAddr) ||
                    addressUseCount.get(remoteAddr) == 0)
                {
                   // if no other sessions are connected to this address, we can
                   // reset the mode
                   connectionModes.remove(remoteAddr);
                }
             }
          }

          return queueInUse;
       }

       /**
        * 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 the leaf node and the router node 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.  On the remote side, if
        * authentication is successful 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.
        * @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 synchronized Session connectVirtual(BootstrapConfig config,
                                                  SessionListener notify)
       throws Exception
       {
          int    port = config.getInt(ConfigItem.SERVER_PORT, -1);
          String host = config.getString(ConfigItem.SERVER_HOST, null);
          InetSocketAddress remoteAddr = new InetSocketAddress(host, port);
          Queue queue = null;
          boolean createdQueue = false;

          synchronized (lock)
          {
             int connectionMode = connectionModes.containsKey(remoteAddr)
                                     ? connectionModes.get(remoteAddr)
                                     : NOT_ESTABLISHED;
             if (connectionMode == DIRECT_MODE)
             {
                throw new IllegalStateException(
                   "Only virtual sessions can be establised to " + remoteAddr +
                   " at this time");
             }
             else
             {
                connectionModes.put(remoteAddr, VIRTUAL_MODE);
             }

             // 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)
          {
             // 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
             int contextID = nextContextID();
             String identity = null;
             synchronized (identityByQueue)
             {
                identity = identityByQueue.get(queue);
             }
             int peerID = callAuthenticateRemote(identity, contextID, queue);

             // registerSession sets a lock on SM.lock, so a lock on queue must be
             // acquired first
             synchronized (queue)
             {
                session = new RequesterSession(queue, null, 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);
                }
                // removed from synchronized section to prevent deadlocks
                if (sessions == null)
                {
                   queue.stop(exc, false);
                }
             }

             exc.fillInStackTrace();

             throw exc;
          }

          return session;
       }

       /**
        * Perform any special initialization which is not handled by the default
        * (common) logic.  This includes checking the bootstrap configuration for
        * valid values and initializing the dispatcher.
        *
        * @throws  ConfigurationException
        *          if any configuration error occurs during initialization.
        */
       protected void init()
       throws ConfigurationException
       {
          connectionModes.clear();

          boolean isSecure = this.config.getBoolean(ConfigItem.SECURE, false);
          int port = -1;
          if (!isSecure)
          {
             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.");
             }
          }
          else
          {
             port = config.getInt(ConfigItem.SERVER_SECURE_PORT, -1);
             if (port <= 0)
             {
                throw new ConfigurationException("Missing or invalid secure port number.");
             }
          }

          int timeout = config.getInt("net", "server", "timeout", 600000);
          String host = config.getString(ConfigItem.SERVER_HOST, null);

          if (host == null || port <= 0 || timeout <= 0)
          {
             throw new ConfigurationException("Missing or incorrect values.");
          }

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