BrokerCore.java
/*
** Module : BrokerCore.java
** Abstract : client side broker implementation.
**
** Copyright (c) 2014-2025, Golden Code Development Corporation.
**
** -#- -I- --Date-- ---------------------------------Description----------------------------------
** 001 MAG 20140707 Implements remote launcher (broker).
** 002 GES 20141228 Modified start() signature to match changes elsewhere.
** 003 EVL 20160223 Javadoc fixes to make compatible with Oracle Java 8 for Solaris 10.
** 004 SBI 20171023 Added getDedicatedHost() and new client parameters: remote:java:args,
** remote:java:classpath, remote:agent:host, remote:agent:user and
** remote:agent:dedicatedMode.
** 005 SBI 20230411 Added severe warning if "localhost" is given instead of client:web:host that
** should be network host or IP address accessible from the server network.
** 006 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 007 GBB 20240709 Hard-coded config names replaced by ConfigItem constants.
** 008 GBB 20240729 Initialize CentralLogger after SecurityManager is created.
** 009 AP 20250303 Added timeout for spawner launching.
*/
/*
** 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.io.*;
import java.lang.management.*;
import java.net.*;
import java.util.*;
import java.util.concurrent.*;
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.*;
/**
* Remote launcher client side implementation.
*/
public class BrokerCore
{
/** Host placeholder */
public static final String PARAM_HOST = "${remote.host}";
/** Secure port placeholder */
public static final String PARAM_PORT = "${remote.secure.port}";
/** Spawner placeholder */
public static final String PARAM_SPAWNER = "${remote.spawner}";
/** JVM arguments placeholder */
public static final String PARAM_JVM = "${remote.jvm.args}";
/** CLASSPATH placeholder */
public static final String PARAM_CP = "${remote.broker.classpath}";
/** Broker P2J server address net:server:host */
public static final String BROKER_HOST = "BROKER_HOST";
/** Broker P2J secured port net:server:secyred_port */
public static final String BROKER_PORT = "BROKER_PORT";
/** Broker runtime JVM arguments */
public static final String BROKER_JVMARGS = "BROKER_JVMARGS";
/** Broker runtime JVM arguments */
public static final String BROKER_JVMARGS_SEPARATOR = "^";
/** Broker runtime class path */
public static final String BROKER_CLASSPATH = "BROKER_CLASSPATH";
/** Logger. */
private static final CentralLogger LOG = CentralLogger.get(BrokerCore.class.getName());
/** JVM arguments and CLASSPATH */
private static List<String> jvmArgs = new LinkedList<>();
/** Export BrokerRemote API */
private static Class<?>[] ifaces = new Class[] { BrokerClientServices.class };
/** Session latch used to notify when terminate */
private static CountDownLatch sessionLatch = new CountDownLatch(1);
/** Assigned UUID */
private static String uuid;
/** P2J Server address */
private static String host;
/** P2J Server secure port */
private static String port;
/** JVM args */
private static String jvmargs;
/** classpath */
private static String classpath;
/** Spawner */
private static String spawner;
/** Launch timeout for the spawner, expressed in seconds */
private static int spawnerLaunchTimeout;
/** Dedicated web host */
private static String dedicatedHost;
/** Associated user */
private static String associatedUser;
/** Dedicated mode */
private static boolean dedicatedMode;
/** Number of retry */
private static int count = 10;
/** Number of seconds to wait until retry */
private static int seconds = 10;
/**
* Start the broker. A number of retry attempts will be made in order to connect to the
* P2J server. Also whenever the connection with server is lost the broker will try
* to automatically reconnect to P2J server.
* <p>
* The number of retries and the interval between retries could be specified in
* configuration (client.xml) file or could be override in command line.
* If missing the default values are used
* Example:
* remote:retry:count=10
* remote:retry:minutes=1
* <p>
* Broker name parameter MUST be override in command line because this parameter
* is used to identify broker clients and to set the name of the logger file.
* Example:
* remote:broker:name=<broker name>
*
* @param cfg
* Bootstrap configuration.
*/
public static void start(final BootstrapConfig cfg)
{
try
{
// get JVM arguments and classpath
initialize(cfg);
@SuppressWarnings("unused")
final SecurityManager secMgr = SecurityManager.createInstance(cfg);
final SessionManager sessMgr = SessionManagerFactory.createLeafNode(cfg);
CentralLogger.setBootstrapConfig(cfg); // after SecurityManager is created
while (count-- > 0)
{
// retry log
LOG.logp(Level.INFO,
"BrokerCore.start()",
"",
CentralLogger.generate("Connecting (%d) ...", count));
connect(sessMgr, cfg);
asleep(seconds);
}
LOG.logp(Level.INFO,
"BrokerCore.start()",
"",
"Exit");
}
catch (Throwable t)
{
LOG.logp(Level.SEVERE,
"BrokerCore.start()",
"",
"Configuration Exception.",
t);
}
}
/**
* This is an exported service used to process remote launch.
*
* @param args
* Remote launch parameters.
*
* @return A <p>BrokerSpawnResult</p> object containing the remote launch result.
*/
public static synchronized BrokerSpawnResult spawn(BrokerSpawnParameters args)
{
// The ProcessBuilder instance
final ProcessBuilder pb = new ProcessBuilder();
// Remote spawn response
final BrokerSpawnResult result = new BrokerSpawnResult();
// Process
Process shell = null;
// Set environment variables.
if (args.getEnvironment() == null)
{
args.setEnvironment(new HashMap<String, String>());
}
// Exports as environment variables
args.getEnvironment().put(BROKER_HOST, host);
args.getEnvironment().put(BROKER_PORT, port);
args.getEnvironment().put(BROKER_JVMARGS, jvmargs);
args.getEnvironment().put(BROKER_CLASSPATH, classpath);
pb.environment().putAll(args.getEnvironment());
// Set directory
File spawnerDir = new File(spawner);
pb.directory(spawnerDir.getAbsoluteFile().getParentFile());
// Command line arguments
List<String> command = new LinkedList<String>();
// Replace place holders
for (String cmd : args.getCommand())
{
// replace placeholders
if (cmd.indexOf(BrokerCore.PARAM_HOST) > -1)
{
command.add(cmd.replace(BrokerCore.PARAM_HOST, host));
}
else if (cmd.indexOf(BrokerCore.PARAM_PORT) > -1)
{
command.add(cmd.replace(BrokerCore.PARAM_PORT, port));
}
else if (BrokerCore.PARAM_SPAWNER.equals(cmd))
{
command.add(spawner);
}
else if (BrokerCore.PARAM_JVM.equals(cmd))
{
command.addAll(jvmArgs);
}
else if (BrokerCore.PARAM_CP.equals(cmd))
{
command.add("-classpath");
command.add(classpath);
}
else
{
command.add(cmd);
}
}
// add client:web:host
command.add("client:web:host=" + dedicatedHost);
command.add("client:web:dedicatedMode=" + String.valueOf(dedicatedMode));
LOG.logp(Level.INFO,
"BrokerCore.spawn()",
"",
CentralLogger.generate("Command line arguments: %s", command.toString()));
pb.command(command);
pb.redirectError(ProcessBuilder.Redirect.INHERIT);
try
{
shell = pb.start();
if (args.isPasswordAuthentication())
{
// write password
OutputStreamWriter os = new OutputStreamWriter(shell.getOutputStream());
os.write(args.getPassword());
os.close();
}
if (!shell.waitFor(spawnerLaunchTimeout, TimeUnit.SECONDS))
{
LOG.log(Level.SEVERE, "Spawner launch procedure exceeded timeout of %d seconds!",
spawnerLaunchTimeout);
result.setExitCode(SpawnError.EXCEEDED_TIMEOUT.getCode());
}
else
{
result.setExitCode(shell.exitValue());
}
LOG.logp(Level.INFO,
"BrokerCore.spawn()",
"",
CentralLogger.generate("Spawn exitValue=%d", result.getExitCode()));
}
catch (Exception e)
{
result.setException(e);
LOG.logp(Level.SEVERE,
"BrokerCore.spawn()",
"",
"Spawning exception.",
e);
}
finally
{
if (shell != null)
{
shell.destroy();
}
}
return result;
}
/**
* This method is used for tests only.
*
* @param uuid
* Assigned UUID
*
* @return UUID parameter.
*/
public static synchronized String ping(String uuid)
{
LOG.logp(Level.INFO,
"BrokerCore.ping()",
"",
CentralLogger.generate("Ping uuid=%s", uuid));
return BrokerCore.uuid;
}
/**
* Returns the dedicated host on which a spawned client will be run.
*
* @return The dedicated host
*/
public static synchronized String getDedicatedHost()
{
return BrokerCore.dedicatedHost;
}
/**
* Get OS loading. Not available on Windows OS -1.00 is always returned.
*
* @return the system load average; or a negative value if not available.
*/
public static synchronized double getSystemLoading()
{
// TODO implements a native solution which works on both Linux and Windows
return ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage();
}
/**
* Get broker JVM arguments and CLASSPATH which is inherited by spawned clients.
* Get other parameters from configuration file.
*
* @param cfg
* Bootstrap configuration.
*/
private static void initialize(BootstrapConfig cfg)
{
cfg.setServer(false);
// UUID
uuid = null;
// get spawner name
host = cfg.getString(ConfigItem.SERVER_HOST, "localhost");
// secured port as string
port = String.valueOf(cfg.getInt(ConfigItem.SERVER_SECURE_PORT, -1));
// number of retry
count = cfg.getInt("remote", "retry", "count", 10);
// retry time in seconds
seconds = cfg.getInt("remote", "retry", "seconds", 10);
// get spawner name
spawner = cfg.getString("remote", "spawner", "file", "./spawn");
// get launch timeout for the spawner
spawnerLaunchTimeout = cfg.getInt(ConfigItem.SPAWNER_LAUNCH_TIMEOUT, ClientSpawner.DEFAULT_TIMEOUT);
// get the dedicated web host
String localHost;
try
{
InetAddress inetAddress = InetAddress.getLocalHost();
localHost = inetAddress.getHostName();
}
catch (UnknownHostException e)
{
localHost = "localhost"; // this localhost should not work for
}
dedicatedHost = cfg.getString("remote", "agent", "host", localHost);
if (Utils.isValidHostNameOrIpAddress(dedicatedHost))
{
LOG.logp(Level.SEVERE,
BrokerCore.class.getName(),
"initialize",
"remote:agent:host should be configured as a network host name or IP address");
}
// get the associated user account name
associatedUser = cfg.getString("remote", "agent", "user", "");
// get the dedicated broker mode
dedicatedMode = cfg.getBoolean("remote", "agent", "dedicatedMode", false);
// JVM arguments
String jargs = cfg.getString("remote", "java", "args", "");
if (jargs.isEmpty())
{
jvmArgs = ManagementFactory.getRuntimeMXBean().getInputArguments();
}
else
{
jvmArgs = Arrays.asList(jargs.split(" "));
}
StringBuilder sb = new StringBuilder();
for (String arg : jvmArgs)
{
if (sb.length() != 0)
{
sb.append(BROKER_JVMARGS_SEPARATOR);
}
sb.append(arg);
}
// store as string
jvmargs = sb.toString();
// classpath
String cp = cfg.getString("remote", "java", "classpath", "");
if (cp.isEmpty())
{
classpath = ManagementFactory.getRuntimeMXBean().getClassPath();
}
else
{
classpath = cp;
}
}
/**
* Sleep for a while.
*
* @param sleepSeconds
* Sleep amount time in seconds.
*/
private static void asleep(long sleepSeconds)
{
try
{
new CountDownLatch(1).await(sleepSeconds, TimeUnit.SECONDS);
}
catch (InterruptedException e)
{
// nothings
}
}
/**
* Establish a secure connection to the P2J server running on the specified port.
* Use the certificate targeted by the specified alias for authentication.
* <p>
* remote:broker:name=<broker name>
*
* @param cfg
* Bootstrap configuration.
*/
private static void connect(SessionManager sessMgr, BootstrapConfig cfg)
{
Session session = null;
try
{
// establish a secure connection with the remote P2J server
session = sessMgr.connectDirect(cfg, null, createSessionListener());
RemoteObject.registerStaticNetworkServer(ifaces, BrokerCore.class);
// Get server exported API
BrokerServerServices remote = (BrokerServerServices) RemoteObject.obtainNetworkInstance(
BrokerServerServices.class,
session);
// register broker
uuid = remote.registerBroker(dedicatedHost, associatedUser, dedicatedMode);
// log a message
LOG.logp(Level.INFO,
"BrokerCore.connect()",
"",
CentralLogger.generate("Broker has been registered uid=%s", uuid));
// after successful register reinitialize retry count.
count = cfg.getInt("remote", "retry", "count", 10);
// wait until session is closed
do
{
remote.start(uuid);
} while (sessionLatch.getCount() > 0);
}
catch (Throwable t)
{
LOG.logp(Level.SEVERE,
"BrokerCore.connect()",
"",
"Connection Exception.",
t);
}
finally
{
cleanup(session);
}
}
/**
* Create a <p>SessionListener</p> object.
*
* @return A <p>SessionListener</p> instance.
*/
private static SessionListener createSessionListener()
{
// session listener
SessionListener sessionListener = new SessionListener()
{
@Override
public void initialize(Session session)
{
sessionLatch = new CountDownLatch(1);
}
@Override
public void terminate(Session session)
{
if (sessionLatch != null && sessionLatch.getCount() == 1)
{
sessionLatch.countDown();
}
}
};
return sessionListener;
}
/**
* Cleanup on exit or error
*/
private static void cleanup(Session session)
{
// close session
if (session != null)
{
session.terminate();
}
// unregister services
String[] list = { ifaces[0].getName() };
RemoteObject.deregisterStaticNetworkServer(list);
}
}