RemoteAppServerConnectionPool.java
/*
** Module : RemoteAppServerConnectionPool.java
** Abstract : A dedicated single-worker pool which will perform the actual request for remote connections.
**
** Copyright (c) 2022-2025, Golden Code Development Corporation.
**
** -#- -I- --Date-- ---------------------------------------Description----------------------------------------
** 001 CA 20220427 First version.
** CA 20221031 Added 'broadcast' to send a message to all workers (like 'disconnect').
** 002 GBB 20240610 LegacyAppServerWork replaced by Consumer<LegacyServiceWorker>.
** 003 GBB 20250403 Removing isSessionFree, since it's no longer determined by the appserver with MSA.
** 004 CA 20250324 Improvements for remote FWD OpenClient connections: ensure that the remote side uses a
** single connection to the FWD server, and also use a client-side pool of worker threads to
** send the requests (the pool size is in sync with the maximum agents available to process
** requests).
*/
/*
** 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.concurrent.atomic.*;
import java.util.logging.*;
import com.goldencode.p2j.cfg.*;
import com.goldencode.p2j.net.*;
import com.goldencode.p2j.security.SecurityManager;
import com.goldencode.p2j.util.*;
import com.goldencode.p2j.util.logging.*;
/**
* When establishing a remote appserver connection, to use the FWD's implementation of the OpenClient, a
* single-worker pool is created. This worker is not shared with other threads, it will establish a private
* appserver connection to the FWD server, while sharing the underlying FWD server physical connection (in
* more specific terms, the {@link AppServerEntry} network proxy will be shared).
* <p>
* In remote connection mode, the target request is executed immediately, but not on this thread - instead,
* will be posted to a thread pool managed by {@link #executor} - this pool size is the same as the maximum
* number of agents available on the remote FWD server, for that appserver.
*/
public class RemoteAppServerConnectionPool
extends AppServerConnectionPool
{
/** Anonymous log instance. */
protected static final CentralLogger LOG = CentralLogger.get(RemoteAppServerConnectionPool.class);
/** The connected FWD servers on this client. */
private static final Map<RemoteServerKey, RemoteServerInfo> connected = new HashMap<>();
/** The pool of workers for a specific FWD server. */
private static final Map<RemoteServerKey, Map<String, ExecutorService>> executors = new HashMap<>();
/** The target appserver name. */
private final String appserver;
/** Flag indicating if the connection is in session-free mode. */
private final boolean sessionFree;
/** The worker which will execute the requests. */
private final RemoteLegacyServiceWorker worker;
/** The target FWD server. */
private final RemoteServerInfo server;
/** The executor for the work to be posted to the FWD server*/
private ExecutorService executor;
/**
* Configure this remote connection for the specified appserver.
*
* @param cfg
* The configuration for the FWD server connection.
* @param appserver
* The appserver name.
*/
public RemoteAppServerConnectionPool(BootstrapConfig cfg, String appserver)
{
this(cfg, appserver, true);
}
/**
* Configure this remote connection for the specified appserver.
*
* @param cfg
* The configuration for the FWD server connection.
* @param appserver
* The appserver name.
* @param sessionFree
* Flag indicating if session-free mode is used.
*/
public RemoteAppServerConnectionPool(BootstrapConfig cfg, String appserver, boolean sessionFree)
{
super("remoteFWDAppServerConnectionPool");
this.appserver = appserver;
this.sessionFree = sessionFree;
RemoteServerKey serverKey = new RemoteServerKey(getServerConfigs(cfg));
this.server = initializeServer(cfg, serverKey, appserver);
this.executor = executors.get(serverKey).get(appserver);
this.worker = new RemoteLegacyServiceWorker(this, server, appserver, true, sessionFree);
}
/**
* Initialize the connection pool - no special code is required for remote connections.
*/
@Override
public void initialize()
{
initialized = true;
// establish the connection now
try
{
execute(() -> worker.getHelper());
}
catch (Throwable t)
{
if (t instanceof RuntimeException)
{
throw (RuntimeException) t;
}
throw new RuntimeException(t);
}
}
/**
* Code to process appserver requests.
* <p>
* In this mode, the request is posted to one of the {@link #executor} threads - this will use the
* {@link #worker}'s appserver connection to the FWD server, and will wait for the request to complete.
*
* @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}.
*/
@Override
public boolean dispatch(String connectionID, LegacyAppServerWork work, String target)
{
try
{
execute(() ->
{
try
{
work.accept(worker);
}
catch (Throwable t)
{
if (t instanceof RuntimeException)
{
throw (RuntimeException) t;
}
throw new RuntimeException(t);
}
});
}
catch (Throwable t)
{
if (t instanceof RuntimeException)
{
throw (RuntimeException) t;
}
throw new RuntimeException(t);
}
return true;
}
/**
* Broadcast a task to all the workers.
*
* @param work
* The work to perform the request.
*/
@Override
public void broadcast(LegacyAppServerWork work)
{
try
{
execute(() -> work.accept(worker));
}
catch (Throwable t)
{
if (t instanceof RuntimeException)
{
throw (RuntimeException) t;
}
throw new RuntimeException(t);
}
}
/**
* For remote connections, the worker has no dedicated thread, as the work is submitted to the shared
* {@link #executor thread pool}, associated with the target {@link #server FWD server}.
*/
@Override
public void terminateWorkers()
{
}
/**
* Get the size of this pool.
*
* @return Always <code>1</code>.
*/
@Override
public int size()
{
return 1;
}
/**
* Get the target appserver name.
*
* @return See above.
*/
@Override
public String getAppserver()
{
return appserver;
}
/**
* Check if the appserver connection must be in session-free mode.
*
* @return See above.
*/
@Override
protected boolean isSessionFree()
{
return sessionFree;
}
/**
* Initialize a connection to the FWD server given in the bootstrap configs. If the FWD server is already
* {@link #connected}, return the {@link RemoteServerInfo details}.
* <p>
* If this is also the first connection to the appserver, then an executor service will be created, to
* manage requests to the appserver, over the single {@link RemoteServerInfo#appServer FWD connection}.
*
* @param cfg
* The bootstrap configuration.
* @param key
* The key for the FWD server connection.
* @param appserver
* The appserver to connect.
*
* @return The details of the remote FWD server.
*/
private static RemoteServerInfo initializeServer(BootstrapConfig cfg,
RemoteServerKey key,
String appserver)
{
RemoteServerInfo server = null;
synchronized (connected)
{
server = connected.get(key);
if (server == null)
{
try
{
// create the SecurityManager
SecurityManager.createInstance(cfg);
// this node will act as a leaf node
SessionManagerFactory.createLeafNode(cfg);
}
catch (NoSuchMethodException |
ConfigurationException e)
{
LOG.log(Level.SEVERE, "Could not establish FWD connection.", e);
return null;
}
server = AppServerHelper.connectToServer(cfg, null, null, null);
if (server.connectError != null)
{
throw new RuntimeException(server.connectError);
}
connected.put(key, server);
server.session.addSessionListener(new SessionListener()
{
@Override
public void terminate(Session session)
{
// cleanup if the FWD server disconnects
connected.remove(key);
Map<String, ExecutorService> services = executors.remove(key);
for (ExecutorService service : services.values())
{
service.shutdown();
}
}
@Override
public void initialize(Session session)
{
}
});
}
Map<String, ExecutorService> appExecutors =
executors.computeIfAbsent(key, (k) -> new ConcurrentHashMap<>());
synchronized (appExecutors)
{
ExecutorService executor = appExecutors.get(appserver);
if (executor == null)
{
int threadNum = server.appServer.getRemotePoolThreads(appserver);
AtomicInteger n = new AtomicInteger();
executor = Executors.newFixedThreadPool(threadNum,
new ThreadFactory()
{
public Thread newThread(Runnable r)
{
Thread t = Executors.defaultThreadFactory().newThread(r);
t.setName("Appserver <" + appserver + "> worker thread #" + n.incrementAndGet());
t.setDaemon(true);
return t;
}
});
appExecutors.put(appserver, executor);
}
}
}
return server;
}
/**
* Get the minimal configurations which specify the details of the FWD server, from the given bootstrap
* config instance.
*
* @param cfg
* The bootstrap configuration.
*
* @return See above.
*/
private static Map<String, String> getServerConfigs(BootstrapConfig cfg)
{
try
{
Map<String, String> res = new HashMap<>();
res.put("net:connection:secure", cfg.getConfigItem("net", "connection", "secure"));
res.put("net:server:host", cfg.getConfigItem("net", "server", "host"));
res.put("net:server:insecure_port", cfg.getConfigItem("net", "server", "insecure_port"));
res.put("net:server:secure_port", cfg.getConfigItem("net", "server", "secure_port"));
res.put("net:keystore:processalias", cfg.getConfigItem("net", "keystore", "processalias"));
return res;
}
catch (ConfigurationException exc)
{
// should not happen, but throw anyway
throw new RuntimeException(exc);
}
}
/**
* Execute the given task and wait the amount of {@link #getTimeout() seconds}. If the timeout is zero,
* then it will wait until the task completes.
*
* @param task
* The task to execute.
*
* @throws ExecutionException
* If the request execution failed.
*/
private void execute(Runnable task) throws ExecutionException
{
int timeout = getTimeout();
try
{
Future future = executor.submit(task);
if (timeout == 0)
{
future.get();
}
else
{
future.get(timeout, TimeUnit.SECONDS);
}
}
catch (ExecutionException ee)
{
LOG.log(Level.SEVERE, "Request execution failed", ee);
throw ee;
}
catch (InterruptedException |
TimeoutException e)
{
LOG.log(Level.WARNING, "Request interrupted after " + timeout + " seconds.");
}
}
/**
* The key with the details for a remote FWD server connection.
*/
private static class RemoteServerKey
{
/** The bootstrap configurations used to establish the remote FWD server connection. */
private final Map<String, String> configs;
/** Pre-calculated hash-code. */
private final int hashCode;
/**
* Create an instance with the specified FWD server configuration.
*
* @param configs
* The configuration for the FWD server connection.
*/
public RemoteServerKey(Map<String, String> configs)
{
this.configs = Collections.unmodifiableMap(new HashMap<>(configs));
this.hashCode = calculateHashCode();
}
/**
* Get the {@link #hashCode}.
*
* @return See above.
*/
@Override
public int hashCode()
{
return hashCode;
}
/**
* Check if the given instance is the same as this one. This means that all the configs in current
* instance must match the configs in the given instance (and vice-versa).
*
* @param obj
* The other instance to compare.
*
* @return <code>true</code> if the instance matches this one.
*/
@Override
public boolean equals(Object obj)
{
if (!(obj instanceof RemoteServerKey))
{
return false;
}
RemoteServerKey key = (RemoteServerKey) obj;
Iterator<Map.Entry<String, String>> iter = this.configs.entrySet().iterator();
while (iter.hasNext())
{
Map.Entry<String, String> e = iter.next();
if (!(key.configs.containsKey(e.getKey()) &&
Objects.equals(e.getValue(), key.configs.get(e.getKey()))))
{
return false;
}
}
return true;
}
/**
* Calculate the hash-code from the given {@link #configs}.
*
* @return See above.
*/
private int calculateHashCode()
{
int result = 17;
result = 37 * result + Integer.hashCode(configs.size());
Iterator<Map.Entry<String, String>> iter = this.configs.entrySet().iterator();
while (iter.hasNext())
{
Map.Entry<String, String> e = iter.next();
result = 37 * result + e.getKey().hashCode();
result = 37 * result + Objects.hashCode(e.getValue());
}
return result;
}
}
}