# Conversation.java

    /*
    ** Module   : Conversation.java
    ** Abstract : represents one side of a two-party single-threaded cooperative
    **            synchronous communication session
    **
    ** Copyright (c) 2007-2023, Golden Code Development Corporation.
    **
    ** -#- -I- --Date-- --JPRM-- -------------------------- Description -----------------------------
    ** 001 GES 20070115   @31830 Created initial version which represents one side of a two-sided
    **                           single-threaded cooperative synchronous communication session.
    ** 002 NVS 20070411   @32927 Conversation threads are transient and as such should call
    **                           dropInitialSecurityContext() before they exit to maintain the
    **                           balance of use for the initial security context.
    ** 003 ECF 20071101   @35878 Refactored net package. Use SessionManager to manage security
    **                           contexts. Other minor cleanup.
    ** 004 ECF 20071127   @36050 Removed remaining SecurityManager reference. NetResource plugin is
    **                           now retrieved from SessionManager. Ensured there is an initial
    **                           security context set before getting plugin.
    ** 005 SVL 20080319   @37556 block() is interrupted if the queue is not running.
    ** 006 ECF 20090205   @42108 Fixed synchronization. The block() method could hold the instance
    **                           monitor for an arbitrarily long time while business logic was
    **                           executed, blocking a call to halt() on the reader thread from
    **                           shutting down the transport after a reader socket error.
    ** 007 GES 20090722   @43332 On server-side, name the thread using context data.
    ** 008 GES 20090723   @43360 Reduced logging of normal session ends to FINE from INFO. Increased
    **                           the usefulness of the entry when generated.
    ** 009 GES 20090816   @43636 Moved context processing into the try block to ensure that cleanup
    **                           happens on failures.
    ** 010 OM  20131014          Added support for notifying session listeners after the correct
    *                            context has been created.
    ** 011 GES 20160419          Reworked synchronization to use an internal lock object which
    **                           eliminates the possibility that external manipulation of the
    **                           instance can interfere with the locking. Fixed a race condition
    **                           where a halt() is called just before the wait() occurs in block()
    **                           causing the wait() to never receive a notify().
    ** 012 GBB 20230512          Logging methods replaced by CentralLogger/ConversionStatus.
    ** 013 GBB 20230620          createThreadName method moved to Utils.
    ** 014 AL2 20250521          Allow interruption of RUNNABLE threads as well to allow terminating sessions
    **                           that seem to be stuck on a repetitive business logic.
    */
    /*
    ** 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.util.logging.*;
    import com.goldencode.p2j.security.*;
    import com.goldencode.p2j.util.*;
    import com.goldencode.p2j.util.logging.CentralLogger;

    /**
     * This class represents one side of a two-sided single-threaded cooperative
     * synchronous communication session.
     * <p>
     * This conversation mode concept has the attribute that when one side of the
     * session is active, the other side is blocked and vice versa.  This makes
     * all processing synchronous and single-threaded.  Control flow transfers
     * back and forth between each side of the session which is why this is
     * described as cooperative.
     * <p>
     * A dedicated thread exists on each side of such a session.  This thread
     * uses a re-entrant blocking function which uses the thread's stack to
     * naturally manages the call/return semantics of each reply to a synchronous
     * request.  This same blocking function is where all inbound synchronous
     * requests are processed (asynchronous requests are still processed in the
     * dispatcher).
     * <p>
     * There are 2 ways of using this class: with a new daemon thread or
     * by capturing an already existing thread.  If you require a new thread
     * to be created for this processing, then use this class as a runnable
     * and then start the thread (set it as daemon mode first).  If you wish to
     * re-use an existing thread, then once the {@link Queue} shifts into
     * conversation mode, the next call into any synchronous message processing
     * will "capture" the thread for the duration of that processing (until the
     * reply message is received).  This is useful in cases where there is an
     * existing "main" thread which calls into a primary entry point and then
     * stays there for the duration of the session.  In this second (capture)
     * approach, it is important to note that the session will not properly
     * respond to synchronous requests until the thread is captured.
     */
    class Conversation
    implements Runnable
    {
       /** Logger. */
       private static final CentralLogger LOG = CentralLogger.get(Conversation.class.getName());

       /** An inbound message to process. */
       private Message msg = null;

       /** Request message if <code>true</code>, otherwise it is a reply. */
       private boolean request = false;

       /** The session-specific network transport. */
       private Queue queue = null;

       /** The singleton instance of the dispatcher. */
       private Dispatcher dispatcher = Dispatcher.getInstance();

       /** Security manager plugin for network resources */
       private NetResource nr = null;

       /** Internal lock mechanism. */
       private Object lock = new Object();

       /** Thread instance on which we are running, must be dedicated to this instance. */
       private Thread worker = null;

       /** Flag to track if the thread is still active. */
       private volatile boolean halted = false;

       /**
        * Constructs an instance.
        *
        * @param    queue
        *           Queue used by the conversation for message passing.
        * @param    worker
        *           In the mode where a new thread is not created for this instance, this is the
        *           thread on which the conversation will be executing.  This MUST be done if we are
        *           in the "capturing" mode where there is an existing thread that will be executing
        *           as the conversation thread.  If a new thread is being created, then {@link #run}
        *           will be called on that thread and this parameter should be <code>null</code>.
        */
       Conversation(Queue queue, Thread worker)
       {
          this.queue = queue;

          if (worker != null)
          {
             this.worker = worker;
          }

          SessionManager sessMgr = SessionManager.get();
          sessMgr.setInitialContext();

          nr = (NetResource) sessMgr.getSecurityPlugin("net");
       }

       /**
        * Implementation of <code>Runnable</code> interface.  Initializes the security context and
        * then calls the main message processing loop for this side of the conversation. This method
        * is NOT expected to ever execute when we are in "capturing" mode.  It only executes when
        * a new thread is being created for the conversation.
        */
       public void run()
       {
          SessionManager sessMgr = SessionManager.get();
          boolean inContext = false;
          boolean apContext = false;

          // this code should not be executed in capturing mode
          assert worker == null;

          // remember our dedicated thread so that we can interrupt processing later
          worker = Thread.currentThread();

          try
          {
             if (sessMgr.isRouter())
             {
                // establish the initial security context (this is only
                // needed here to allow the security cache to migrate each
                // thread to any newer initial context that may be set over time
                sessMgr.setInitialContext();
                inContext = true;

                // set up the security context which is valid for the life of this thread
                sessMgr.setContext(queue);
                apContext = true;

                // name the new thread so thread dumps are more meaningful
                String name = Utils.createThreadName("Conversation");
                Thread.currentThread().setName(name);

                // force better naming for the protocol threads (must be done here
                // from a thread that has the proper security context)
                queue.setProtocolNames();

                // allows listeners to perform any required tasks on the proper context AFTER it
                // has been created and before client code will be executed
                ((BaseSession)sessMgr.getSession()).sendInitializationEvent();
                // cast is safe, value returned by getSession() is internally stored as BaseSession
             }

             // clear traces of interrupted status, if any
             Thread.interrupted();

             block();
          }
          catch (ApplicationRequestedStop exc)
          {
             // normal termination of the queue
             if (LOG.isLoggable(Level.FINE))
             {
                LOG.logp(Level.FINE, "Conversation.run()", "", exc.getMessage());
             }
          }
          catch (Exception exc)
          {
             // log the error and exit the thread
             if (!(exc instanceof InterruptedException))
             {
                if (LOG.isLoggable(Level.WARNING))
                {
                   LOG.log(Level.WARNING,
                           "--- exception in Conversation.run() ---",
                           exc);
                }
             }
          }

          finally
          {
             if (apContext)
             {
                sessMgr.restoreContext();
             }

             if (inContext)
             {
                // drop security context
                sessMgr.dropInitialContext();
             }
          }
       }

       /**
        * Halt the conversation.
        */
       void halt()
       {
          synchronized (lock)
          {
             halted = true;
             msg    = null;
             lock.notify();
          }

          if (worker != null)
          {
             // one last attempt to unblock a conversation thread
             worker.interrupt();
          }
       }

       /**
        * Notify the conversation that a request message has been received.
        *
        * @param    msg
        *           The incoming message.
        */
       void request(Message msg)
       {
          synchronized (lock)
          {
             this.msg     = msg;
             this.request = true;
             lock.notify();
          }
       }

       /**
        * Notify the conversation that a reply message has been received.
        *
        * @param    msg
        *           The incoming message.
        */
       void reply(Message msg)
       {
          synchronized (lock)
          {
             this.msg     = msg;
             this.request = false;
             lock.notify();
          }
       }

       /**
        * Block until a reply message is received where the request ID matches
        * the given ID.
        *
        * @param    reqId
        *           The request ID to wait for.
        *
        * @return   The reply message.
        */
       Message waitMessage(int reqId)
       throws InterruptedException,
              RestrictedUseException,
              IllegalAccessException,
              NoSuchMethodException,
              ApplicationRequestedStop,
              ProtocolViolation
       {
          Message response = block();
          int replyId = response.getRequestID();

          if (replyId != reqId)
          {
             String err = "Expected ID " + reqId + " but rec'd ID " + replyId;
             throw new ProtocolViolation(err);
          }

          return response;
       }

       /**
        * This is the main message processing loop for this side of the
        * conversation.  The method is designed to be re-entrant to allow the
        * thread's stack to naturally handle the synchronous nature/state of
        * transaction support.
        * <p>
        * This method will not return until it receives a reply message.
        *
        * @return   The reply ID of the message just received.
        */
       private Message block()
       throws InterruptedException,
              RestrictedUseException,
              IllegalAccessException,
              NoSuchMethodException,
              ApplicationRequestedStop
       {
          Message msg = null;
          boolean request = false;

          while (queue != null && queue.isRunning())
          {
             synchronized (lock)
             {
                // check if the reply has already arrived, if we wait in that case then the notify
                // will never occur because that has already been processed; the same thing can
                // occur if halt() has been called in between the check for halting below and the
                // top of the loop, thus we check again here to ensure that we don't wait() when
                // there will never be another notify()
                if (this.msg == null && !halted)
                {
                   try
                   {
                      lock.wait(1000);
                   }
                   catch(InterruptedException ignore)
                   {
                   }
                }

                // check for a halt
                if (halted)
                {
                   throw new ApplicationRequestedStop();
                }

                if (this.msg == null)
                {
                   continue;
                }
                msg     = this.msg;
                request = this.request;

                // reset the state of the conversation to allow for blocking
                this.msg     = null;
                this.request = false;
             }

             if (request)
             {
                // this is an incoming request for service
                InboundRequest ir = new InboundRequest(msg, queue);
                dispatcher.processInbound(ir, false, nr);
             }
             else
             {
                // this must be a reply we are waiting for
                break;
             }
          }

          if (queue == null || !queue.isRunning())
          {
             throw new ApplicationRequestedStop();
          }

          return msg;
       }
    }