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