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;
}
}