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