# BaseSession.java

    /*
    ** Module   : BaseSession.java
    ** Abstract : Abstract implementation of Session interface
    **
    ** Copyright (c) 2004-2025, Golden Code Development Corporation.
    **
    ** -#- -I- --Date-- --JPRM-- ---------------------------- Description ----------------------------
    ** 001 ECF 20071101  @35861  Created initial version. Abstract
    **                           implementation of Session interface.
    ** 002 CA  20110201          Track the session termination state, so that
    **                           isRunning() returns the correct value, regardless
    **                           of the queue's state.
    ** 003 CA  20111129          Added getter for the security context field.
    ** 004 OM  20131018          Added sendInitializationEvent() trigger for SessionListener interface.
    ** 005 CA  20140902          Added addSessionListener(SessionListener, boolean) to conditionally
    **                           register the listener, based on the queue state (running or not), and
    **                           also ensuring the queue state is not changed during listener
    **                           registration.
    ** 006 GBB 20230512          Logging methods replaced by CentralLogger/ConversionStatus.
    ** 007 GBB 20230708          SecurityManager context & session methods calls updated.
    ** 008 CA  20250321          Changed the listeners to a linked set - they need to be processed in the
    ***                          registration order; also, use a copy for event processing.
    */
    /*
    ** 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.util.*;
    import java.util.logging.*;
    import com.goldencode.p2j.security.*;
    import com.goldencode.p2j.security.SecurityManager;
    import com.goldencode.p2j.util.logging.*;

    /**
     * Abstract implementation of the <code>Session</code> interface. Stores a
     * queue, a security context, and a numeric value which uniquely identifies
     * that context within the local network node.  A <code>null</code> context
     * and a context ID of 0 are valid;  these values indicate there is no
     * security context for the session.
     * <p>
     * All methods of <code>Session</code>'s superinterface, {@link
     * LowLevelSession}, are delegated to the queue.  The remaining methods
     * generally are called by, and call, the {@link SessionManager}.
     */
    abstract class BaseSession
    implements Session,
               MessageTypes
    {
       /** Logger */
       protected static final CentralLogger LOG =
          CentralLogger.get(Session.class.getName());

       /**
        * List of objects to receive event notifications.  The listeners are processed in the registration order.
        */
       private final Set<SessionListener> listeners = new LinkedHashSet<SessionListener>();

       /** Message transport queue */
       private final Queue queue;

       /** Security context identifier */
       private final int contextID;

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

       /** Track this session's terminated state. */
       private boolean terminated = false;

       /**
        * Constructor.
        *
        * @param   queue
        *          Message transport.
        * @param   context
        *          Security context.
        * @param   contextID
        *          Security context ID.
        */
       protected BaseSession(Queue queue, Object context, int contextID)
       {
          this.queue = queue;
          this.context = context;
          this.contextID = contextID;
       }

       /**
        * Add a new instance into the list of objects which will be notified
        * of session events.
        *
        * @param    listener
        *           The listener to add.
        */
       public void addSessionListener(SessionListener listener)
       {
          if (listener == null)
          {
             return;
          }

          synchronized (listeners)
          {
             if (!listeners.contains(listener))
             {
                listeners.add(listener);
             }
          }
       }

       /**
        * Add a new instance into the list of objects which will be notified
        * of session events.
        * <p>
        * If the <code>check</code> flag is set, then it ensure that the session is still connected
        * before adding the listener.  If the session is not connected, then the listener will not
        * be added and this will return <code>false</code>.
        *
        * @param   listener
        *          The listener to add.
        * @param   check
        *          Flag to check session state before adding.
        *
        * @return  <code>true</code> if the listener was added. <code>false</code> otherwise.
        */
       public boolean addSessionListener(SessionListener listener, boolean check)
       {
          if (!check)
          {
             addSessionListener(listener);

             return true;
          }

          // if the flag is set, check if the session is still running, and if so, add the listener.
          // this needs to be synchronized to ensure that the session state and the add session
          // listener code are atomic.
          synchronized (queue)
          {
             if (!isRunning())
             {
                return false;
             }

             addSessionListener(listener);

             return true;
          }
       }

       /**
        * Remove the listener from the list of objects to be notified about
        * session events.
        *
        * @param    listener
        *           The listener to remove.
        */
       public void removeSessionListener(SessionListener listener)
       {
          if (listener == null)
          {
             return;
          }

          synchronized (listeners)
          {
             listeners.remove(listener);
          }
       }

       /**
        * Sets the instance of the state synchronizer the reader and writer
        * threads should use, if and only if there is no other state synchronizer
        * already in use.
        *
        * @param   sync
        *          The synchronizer to register or <code>null</code> if the
        *          currently registered synchronizer should be deregistered.
        */
       public void registerSynchronizer(StateSynchronizer sync)
       {
          queue.registerSynchronizer(sync);
       }

       /**
        * Terminate the network session.  This method notifies all session
        * listeners of the termination event, and will stop the underlying queue
        * if this is the only session using it.
        */
       public void terminate()
       {
          SessionManager.get().endSession(this, true, true);
       }

       /**
        * Preprocess the given message and then send it as a synchronous request
        * and wait for reply.  Return message payload as a result.
        * <p>
        * The message passed as a parameter must be completely filled with all
        * required information - routing key should point to valid entry point. If
        * addresses in the routing key are not set then destination is assumed at
        * directly connected node and appropriate addresses are assigned
        * automatically.
        * <p>
        * Caller can specify timeout for waiting reply message. If timeout is set
        * to 0 then queue will wait for reply as long as queue is running.
        *
        * @param   message
        *          Request message.
        * @param   timeout
        *          Reply wait timeout.
        *
        * @return  Reply message payload.
        *
        * @throws  Exception
        *          if reply received from remote side has an exception as its
        *          payload, or if a timeout occurs before the reply message is
        *          received, or if a problem occurs during transmission.
        */
       public Object transact(Message message, int timeout)
       throws Exception
       {
          preprocess(message);

          return queue.transact(message, timeout);
       }

       /**
        * Preprocess the given message and then send it as an asynchronous request
        * without waiting for a reply.  The message type will be forced to
        * <code>REQUEST_ASYNCH</code>.
        *
        * @param   message
        *          Request message.
        */
       public void forward(Message message)
       {
          preprocess(message);
          queue.forward(message);
       }

       /**
        * Send an echo request message using provided data as a message payload
        * and wait for reply. Return message payload as a result.
        * <p>
        * If instance of <code>RoutingKey</code> is provided then message is
        * sent to node pointed by the <code>RoutungKey</code>. Otherwise
        * directly connected node is assumed. Note that processing of the ECHO
        * message ignores entry point information in <code>RoutingKey</code> so
        * an instance of <b>RoutingKey </b> created with default constructor can
        * be used.
        * <p>
        * Caller can specify timeout for waiting reply message. If this is not
        * necessary then 0 should be passed as If timeout is set to 0 then queue
        * will wait for reply as long as queue is running.
        *
        * @param    obj
        *           Message payload.
        * @param    timeout
        *           Reply wait timeout.
        * @param    key
        *           The destination node address information.
        *
        * @return   Payload of the echo reply.
        *
        * @throws   InterruptedException
        *           forwarded from transactImpl()
        * @throws   RequestTimeoutException
        *           forwarded from transactImpl()
        * @throws   ProtocolViolation
        *           in case of protocol errors.
        * @throws   Exception
        *           Is thrown if a timeout occurs before the reply message
        *           is received or if a problem occurs during transmission.
        */
       public Object echo(Serializable obj, int timeout, RoutingKey key)
       throws ProtocolViolation,
              InterruptedException,
              RequestTimeoutException,
              Exception
       {
          Message message = new Message(ECHO, obj);

          if (key == null)
          {
             key = new RoutingKey();
             key.setSourceAddress(queue.getNodeAddress());
             key.setDestinationAddress(queue.getRemoteAddress());
          }
          message.setKey(key);

          Message reply = queue.transactImpl(message, timeout);

          if (reply == null)
             return null;

          if (reply.getType() == ECHO_REPLY)
             return reply.getPayload();

          throw new ProtocolViolation("Reply is not ECHO_REPLY");
       }

       /**
        * Get the node address assigned to the session.
        *
        * @return  Node address.
        */
       public int getNodeAddress()
       {
          return queue.getNodeAddress();
       }

       /**
        * Get the remote node address known to the session.
        *
        * @return  Remote node address.
        */
       public int getRemoteAddress()
       {
          return queue.getRemoteAddress();
       }

       /**
        * Returns the operational state of this session.
        *
        * @return  <code>true</code> if the session is running (operational)
        *          and <code>false</code> if the session is closed.
        */
       public boolean isRunning()
       {
          return !terminated && queue.isRunning();
       }

       /**
        * Return the exception which was thrown for an exceptional termination of
        * the session, if any.
        * <p>
        * If not <code>null</code>, it will contain the queue termination reason.
        *
        * @return  The session termination reason.  This will be <code>null</code>
        *          if the session is not yet terminated or if the session's
        *          termination was not abnormal.
        */
       public Exception getException()
       {
          return queue.getException();
       }

       /**
        * Preprocess a message before it is delivered.
        *
        * @param   message
        *          Message to be processed.
        */
       protected abstract void preprocess(Message message);

       /**
        * Terminate the current session with the security manager.  Subclasses
        * which must leave the context intact (because it was not created with the
        * session) must override this method.
        */
       protected void cleanupContext()
       {
          try
          {
             // Allow the security manager to clean up its resources (it will
             // remove the context BUT it does not kill any threads associated
             // with that context, that must be handled by the queue).
             SecurityManager.getInstance().sessionSm.terminateSession(context);
          }
          catch (Exception exc)
          {
             // Log that there was an unexpected failure
             if (LOG.isLoggable(Level.SEVERE))
             {
                LOG.logp(Level.SEVERE,
                         "Session.stop()",
                         "",
                         "Abend during session cleanup.",
                         exc);
             }
          }
          finally
          {
             context = null;
          }
       }

       /**
        * Get the session's queue.
        *
        * @return  Queue.
        */
       Queue getQueue()
       {
          return queue;
       }

       /**
        * Get the session's security context ID.
        *
        * @return  Context ID.
        */
       int getContextID()
       {
          return contextID;
       }

       /**
        * End the network session.  This involves terminating the security
        * session with the security manager.
        */
       final void end()
       {
          if (context != null)
          {
             cleanupContext();
          }

          terminated = true;
       }

       /**
        * Respond to an echo request by simply changing the message type to
        * <code>ECHO_REPLY</code> and sending it back to its source node.
        *
        * @param   msg
        *          Echo message.
        */
       void echoReply(Message msg)
       {
          msg.setType(ECHO_REPLY);
          queue.enqueueOutbound(msg);
       }

       /**
        * Associate the session's security context with the current thread.
        */
       void setContext()
       {
          try
          {
             SecurityManager.getInstance().contextSm.pushAndSwitchSecurityContext(context);
          }
          catch (RestrictedUseException ru)
          {
             throw new RuntimeException("Invalid security context switch");
          }
       }

       /**
        * Send notification to all registered nodes that queue is about to start and have to be
        * initialized.
        */
       void sendInitializationEvent()
       {
          synchronized (listeners)
          {
             // use a copy, the listener can remove itself from the queue
             for (SessionListener listen : new ArrayList<>(listeners))
             {
                listen.initialize(this);
             }
          }
       }

       /**
        * Send notification to all registered nodes that queue is about to terminate.
        */
       void sendTerminationEvent()
       {
          synchronized (listeners)
          {
             // use a copy, the listener can remove itself from the queue
             for (SessionListener listen : new ArrayList<>(listeners))
             {
                listen.terminate(this);
             }
          }
       }

       /**
        * Get the security context for this session.
        *
        * @return   See above.
        */
       Object getContext()
       {
          return context;
       }
    }