AppServerConnectionPool.java
/*
** Module : AppServerConnectionPool.java
** Abstract : Provides a pool of workers to which the tasks received by the SOAP, REST or java appserver
** clients are delegated.
**
** Copyright (c) 2021-2025, Golden Code Development Corporation.
**
** -#- -I- --Date-- -------------------------------------Description------------------------------------------
** 001 CA 20211214 Created initial version.
** CA 20211216 Improved initialization of the worker threads - now the pool waits for all workers to have
** at least one appserver initialization for each one.
** CA 20220105 Compute the timeout early, as the service registration requires it when they are built.
** CA 20220114 Changed sessionId to long. Improvement for worker startup.
** CA 20220208 Mark pools which are used for the web services managed by FWD.
** CA 20220427 Refactored to allow remote OpenClient-style connections, from outside the FWD server.
** CA 20221031 Added 'broadcast' to send a message to all workers (like 'disconnect').
** 002 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 003 GBB 20230825 SecurityManager context & certificate methods calls updated.
** 004 EVL 20231124 Decreased sleep time to 50ms to improve responce time.
** 005 GBB 20240610 LegacyAppServerWork replaced by Consumer<LegacyServiceWorker>. Setting client type.
** 006 CA 20250106 Unless a request is bound to a worker (like an internal entry to a persistent proc or
** broadcast), workers can otherwise use a poll method to get tasks, and not push. Push is
** used only when a task *must* reach a certain worker.
** 007 GBB 20250124 doInitialize to wait for just one agent before continuing with init.
** 008 GBB 20250403 Support for the new multi-session appserver.
** 009 CA 20250324 Extracted the code waiting for appserver initialization to
** 'AppServerManager.waitForAppserver'.
*/
/*
** 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.main;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.*;
import java.util.logging.*;
import com.goldencode.p2j.cfg.*;
import com.goldencode.p2j.directory.Base64;
import com.goldencode.p2j.main.ServerKeyStore.Store;
import com.goldencode.p2j.net.*;
import com.goldencode.p2j.security.*;
import com.goldencode.p2j.security.SecurityManager;
import com.goldencode.p2j.util.*;
import com.goldencode.p2j.util.appserver.*;
import com.goldencode.p2j.util.logging.*;
import com.goldencode.util.RandomWordGenerator;
/**
* Provides a pool of {@link LegacyServiceWorker workers}, which have a FWD context as configured in the
* client's type (directory's soap, rest, web app node).
* <p>
* The workers are automatically connected when the first task is received. The workers should have a 1-to-1
* mapping between them and the agents, and the pool size should never be configured larger than the number
* of agents available.
*/
public class AppServerConnectionPool
{
/** Flag indicating if the pool has initialized. */
protected volatile boolean initialized = false;
/** Anonymous log instance. */
private static final CentralLogger LOG = CentralLogger.get(AppServerConnectionPool.class.getName());
/** The pool of workers dedicated to executing legacy services. */
private final List<LegacyServiceWorker> workers = new ArrayList<>();
/** Common task queue on which the workers are polling. */
private final BlockingQueue<Consumer<LegacyServiceWorker>> tasks = new LinkedBlockingQueue<>();
/** Mapping of workers by their connection ID. */
private final Map<String, LegacyServiceWorker> workersByConnection = new ConcurrentHashMap<>();
/** The directory node from where to read the configuration (rest, soap, WebHandler, someWebApp). */
private final String type;
/** Flag indicating if this worker pool is for web services managed by FWD. */
private final LegacyServiceHandler.ServiceType forWebServiceType;
/** The appserver definition. */
private AppserverDefinition def;
/** Flag for the type of the appserver, new multi-session agent appserver. */
private boolean isNewMultiSession;
/** The timeout for a service call, defaults to 10 seconds. */
private int timeout;
/** The max number of connections for the appserver. */
private int maxConnections;
/** The number of {@link #workers} started in this pool. */
private int poolSize;
/** If the pool of workers should be in Session-managed session model. */
private boolean isPoolSessionManaged;
/** The session ID generator. */
private long nextSessionId = 0;
/**
* Create a new pool.
*
* @param webServiceType
* The legacy web service type.
*/
public AppServerConnectionPool(LegacyServiceHandler.ServiceType webServiceType)
{
this(webServiceType.toString(), webServiceType);
}
/**
* Create a new pool.
*
* @param name
* The directory node from where to read the configuration.
*/
public AppServerConnectionPool(String name)
{
this(name, null);
}
/**
* Private constructor.
*
* @param name
* The directory node from where to read the configuration.
* @param webServiceType
* The legacy web service type.
*/
private AppServerConnectionPool(String name, LegacyServiceHandler.ServiceType webServiceType)
{
this.type = name;
this.forWebServiceType = webServiceType;
}
/**
* Get the next session ID.
*
* @return See above.
*/
public synchronized long nextSessionId()
{
return nextSessionId++;
}
/**
* Check if this pool is for web services ran from within FWD.
*
* @return See above.
*/
public boolean isWebService()
{
return forWebServiceType != null;
}
/**
* Get the state of the {@link #initialized} flag.
*
* @return See above.
*/
public boolean isInitialized()
{
return initialized;
}
/**
* Broadcast a task to all the workers.
*
* @param work
* The work to perform the request.
*/
public void broadcast(LegacyAppServerWork work)
{
for (int i = 0; i < workers.size(); i++)
{
LegacyServiceWorker worker = workers.get(i);
worker.addWork(work);
}
}
/**
* Code to process appserver requests.
*
* @param connectionID
* When not-null, force to use the worker with the specified connection ID.
* @param work
* The work to perform the request.
* @param target
* The target path.
*
* @return <code>true</code> if the request was processed by a {@link #workers worker}.
*/
public boolean dispatch(String connectionID, LegacyAppServerWork work, String target)
{
// TODO: increase the workers based on the appserver settings (max_agents, etc)????
boolean toPoolQueue = connectionID == null;
// find a worker
LegacyServiceWorker worker;
if (toPoolQueue)
{
// this can be pushed to the global queue; the worker is not calculated here, any worker can take it
worker = null;
}
else
{
// this needs to be posted directly at the worker for this connection
worker = workersByConnection.get(connectionID);
if (worker == null)
{
// work interrupted
LOG.log(Level.SEVERE,
"Could not find worker for connection " + connectionID + " when processing " + target);
return false;
}
}
Throwable[] exc = new Throwable[1];
CountDownLatch down = new CountDownLatch(1);
Consumer<LegacyServiceWorker> theWork = (w) ->
{
ClientType[] initialClientTypes = SessionUtils.getClientTypes();
SessionUtils.setClientTypes(work.getClientType());
try
{
work.accept(w);
}
catch (Throwable t)
{
exc[0] = t;
}
finally
{
SessionUtils.setClientTypes(initialClientTypes);
down.countDown();
}
};
if (toPoolQueue)
{
tasks.add(theWork);
for (int i = 0; i < workers.size(); i++)
{
LegacyServiceWorker w = workers.get(i);
w.notifyTask();
}
}
else
{
worker.addWork(theWork);
}
boolean done = false;
try
{
if (timeout == 0)
{
down.await();
done = true;
}
else
{
done = down.await(timeout, TimeUnit.SECONDS);
}
}
catch (InterruptedException e)
{
// work interrupted
LOG.log(Level.SEVERE, "Interrupted worker for " + target);
return false;
}
if (exc[0] != null)
{
if (exc[0] instanceof RuntimeException)
{
throw (RuntimeException) exc[0];
}
throw new RuntimeException(exc[0]);
}
if (!done)
{
LOG.log(Level.SEVERE, type + " call " + target + " terminated, timeout elapsed.");
return false;
}
return true;
}
/**
* Terminate the service worker threads.
*/
public void terminateWorkers()
{
// TODO: should this call Disconnect procedure for the connection of each worker? note that for
// State-free appservers (which are used for SOAP, REST, WebHandler and some multi-session appserver
// clients) does not support Disconnect procedure.
synchronized (this)
{
for (LegacyServiceWorker worker : workers)
{
try
{
worker.stop();
}
catch (Exception exc)
{
LOG.logp(Level.SEVERE,
"LegacyServiceHandler.terminate()",
"",
"Could not shutdown the " + type + " workers!",
exc);
}
}
}
}
/**
* Initialize the connection pool, in a separate thread.
*/
public void initialize()
{
String appserverName = Utils.getDirectoryNodeString(null, type + "/appserver", null, false);
ClassicAppserverDefinition classicDef = ClassicAppserverDefinition.get(appserverName);
if (classicDef == null)
{
MultiSessionAppserverDefinition msDef = MultiSessionAppserverDefinition.get(appserverName);
if (msDef == null)
{
LOG.log(Level.WARNING,
"Trying to initialize connection pool '%s' for appserver '%s', but no definition found.",
type,
appserverName);
return;
}
this.def = msDef;
this.isNewMultiSession = true;
this.maxConnections = msDef.getMaxConnectionsPerAgent() * msDef.getMaxAgents();
// this is already used in the MSA server
this.timeout = msDef.getSessionExecutionTimeLimit();
}
else
{
this.def = classicDef;
this.isNewMultiSession = false;
this.maxConnections = classicDef.getMaxAgents();
Integer timeout = classicDef.getRequestTimeout();
this.timeout = timeout == null ? 10 : timeout;
}
Thread t = new AssociatedThread(this::doInitialize);
t.setName("Worker starter for " + type);
t.start();
}
/**
* Initialize this pool.
*/
private void doInitialize()
{
try
{
// wait for the FWD server to start.
SessionManager.get().waitUntilReady();
}
catch (InterruptedException e1)
{
LOG.log(Level.FINE, "", e1);
return;
}
SecurityManager sm = SecurityManager.getInstance();
// these must be read from the server's context
String alias = Utils.getDirectoryNodeString(null, type + "/alias", null, false);
String ksfile = Utils.getDirectoryNodeString(null, type + "/keyStore", null, false);
byte[] ksbytes;
String kspass;
String kepass;
if (ksfile == null)
{
kspass = RandomWordGenerator.create();
kepass = RandomWordGenerator.create();
ksbytes = sm.certificateSm.getEncryptedKeyStore()
.getEncryptedKeyStore(alias, kspass, kepass, Store.TRUST_STORE);
}
else
{
kspass = Utils.getDirectoryNodeString(null, type + "/storePassword", null, false);
kepass = Utils.getDirectoryNodeString(null, type + "/entryPassword", null, false);
ksbytes = null;
}
this.poolSize = Utils.getDirectoryNodeInt(null, type + "/poolSize", maxConnections, false);
this.isPoolSessionManaged = Utils.getDirectoryNodeBoolean(null,
type + "/isPoolSessionManaged",
false,
false);
if (maxConnections < poolSize)
{
LOG.log(Level.WARNING,
"The number of max appserver sessions " + maxConnections +
" is less than the number of configured " + type + " workers " + workers.size() + ".");
}
AppServerManager.waitForAppserver(def.getAppserverName());
CountDownLatch counter = new CountDownLatch(poolSize);
for (int i = 0; i < poolSize; i++)
{
Thread t = new Thread(() ->
{
try
{
BootstrapConfig cfg = new BootstrapConfig();
cfg.setConfigItem("net", "connection", "secure", "true");
cfg.setConfigItem("security", "keystore", "processalias", alias);
if (ksfile != null)
{
cfg.setConfigItem("security", "keystore", "filename", ksfile);
cfg.setConfigItem("access", "password", "keystore", kspass);
cfg.setConfigItem("access", "password", "keyentry", kepass);
}
else
{
cfg.setConfigItem("security", "keystore", "bytes",
Base64.byteArrayToBase64(ksbytes));
cfg.setConfigItem("access", "password", "keystore", kspass);
}
Object context = sm.authenticateServer(cfg);
if (context == null)
{
LOG.logp(Level.SEVERE,
"Could not authenticate the worker for the " + type + " services!",
"",
"!");
return;
}
}
catch (ConfigurationException | RestrictedUseException e)
{
LOG.severe("Could not perform the " + type + " services authentication!", e);
return;
}
// mark this as 'headless'
SecurityManager.getInstance().contextSm.setHeadless();
boolean isSessionFree = isSessionFree();
LegacyServiceWorker worker = new LegacyServiceWorker(this,
def.getAppserverName(),
def.isMultiSession(),
isSessionFree,
forWebServiceType);
workers.add(worker);
counter.countDown();
worker.run();
});
t.setName(type.toUpperCase() + " worker #" + i);
t.setDaemon(true);
t.start();
}
// wait for all threads to start and be added to the pool.
try
{
// wait 1 seconds for each worker to start
counter.await(poolSize, TimeUnit.SECONDS);
int sessions = AppServerManager.getNumberOfSessions(def.getAppserverName());
initialized = sessions > 0;
if (!initialized)
{
LOG.log(Level.SEVERE,
"Could not initialize the workers for pool " + type + " - no appserver agents started.");
}
else if (sessions < poolSize)
{
int started = poolSize - (int) counter.getCount();
LOG.log(Level.WARNING, "Initialized only %d agents in a pool of size %d workers for %s.",
sessions, started, type);
}
}
catch (InterruptedException e)
{
// ignore
}
}
/**
* Get the service handler's {@link #type}.
*
* @return The service type.
*/
public String getType()
{
return type;
}
/**
* Get the {@link #timeout}.
*
* @return See above.
*/
public int getTimeout()
{
return timeout;
}
/**
* Set the timeout for a request to complete.
*
* @param timeout
* The timeout for a request to complete (in seconds). Use zero to wait indefinitely.
*/
public void setTimeout(int timeout)
{
this.timeout = timeout;
}
/**
* Get the size of the pool.
*
* @return The pool {@link #poolSize size}.
*/
public int size()
{
return poolSize;
}
/**
* Get the target appserver name.
*
* @return See above.
*/
public String getAppserver()
{
return def.getAppserverName();
}
/**
* Get an available task from the pool's {@link #tasks}.
*
* @return An available task (or <code>null</code> if {@link #tasks the queue} is empty).
*/
public Consumer<LegacyServiceWorker> getTask()
{
return tasks.poll();
}
/**
* Register the worker with the specified connection ID.
*
* @param connectionID
* The connection ID.
* @param worker
* The service worker.
*/
protected void registerWorker(String connectionID, LegacyServiceWorker worker)
{
workersByConnection.put(connectionID, worker);
}
/**
* Check if the appserver connection must be in session-free mode.
*
* @return See above.
*/
protected boolean isSessionFree()
{
return isNewMultiSession ?
forWebServiceType == LegacyServiceHandler.ServiceType.REST ||
forWebServiceType == LegacyServiceHandler.ServiceType.WEB ||
!isPoolSessionManaged :
((ClassicAppserverDefinition) def).isSessionFree();
}
}