LegacyServiceWorker.java
/*
** Module : LegacyServiceWorker.java
** Abstract : A worker to process legacy service requests.
**
** Copyright (c) 2019-2025, Golden Code Development Corporation.
**
** -#- -I- --Date-- ---------------------------------------Description----------------------------------------
** 001 CA 20190614 First version.
** 002 CA 20190628 Proper SINGLETON and SINGLE-RUN implementation; misc improvements.
** 003 CA 20190701 Renamed from RestWorker to LegacyServiceWorker.
** 004 CA 20200514 Added support for SOAP web services.
** 005 CA 20211112 The worker must be aware if the appserver is in MSA mode (for error processing and other
** differences between classic and multisession).
** CA 20211214 Integrated the workers with the legacy java clients. Misc fixes related to State-free
** connections and agents bound to remote procedures (the binding must be made using the
** agent's ID).
** CA 20211216 Expose the task counter for each worker with a JMX bean.
** CA 20220114 Changed sessionId to long.
** CA 20220208 For calls originating from outside the FWD server, do not serialize the legacy OO errors.
** 006 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 007 HC 20240222 Enabled JMX on FWD Client.
** 008 LS 20240912 Changed getProcedures to return a copy of the set and not the actual reference.
** 009 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.
** 010 GBB 20250403 Sending the service type to connect, leaving the appserver to identify the session model.
** 011 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.function.*;
import java.util.logging.*;
import com.goldencode.p2j.jmx.*;
import com.goldencode.p2j.net.SessionManager;
import com.goldencode.p2j.util.*;
import com.goldencode.p2j.util.logging.*;
/**
* Worker class to process the requests.
*/
public class LegacyServiceWorker
implements Runnable
{
/** Anonymous log instance. */
private static final CentralLogger LOG = CentralLogger.get(LegacyServiceWorker.class.getName());
/** Flag indicating if the task worker is running. */
private volatile boolean running = false;
/** A counter of all tasks done by all workers. */
private static final ValuesCounter TASK_COUNTER =
ValuesCounter.getInstance(FwdServerJMX.MapCounter.LegacyServiceWorkerTaskCounter);
/** The connection to the remote appserver executing the requests. */
protected AppServerHelper helper = null;
/** The appserver name. */
protected final String appserver;
/** Flag indicating if the target appserver is in Session-free mode. */
protected final boolean sessionFree;
/** Flag indicating if the target is a multi-session agent appserver. */
private final boolean multiSession;
/** The tasks to be ran. */
private final LinkedList<Consumer<LegacyServiceWorker>> tasks = new LinkedList<>();
/** The type of web service the worker is used for. */
private final LegacyServiceHandler.ServiceType forWebServiceType;
/** The service pool. */
private AppServerConnectionPool pool;
/**
* The remote procedures currently created by this worker, per each session ID. When the client session
* disconnects, this will be automatically deleted.
*/
private final Map<Long, Set<ProxyProcedureWrapper>> remoteProcedures = new ConcurrentHashMap<>();
/** The thread name where this instance runs. */
private String threadName = null;
/**
* Initialize this worker.
*
* @param pool
* The service pool.
* @param appserver
* The appserver name.
* @param multiSession
* Flag indicating if it's a multi-session agent appserver.
* @param sessionFree
* Flag indicating if the appserver is Session-free.
*/
public LegacyServiceWorker(AppServerConnectionPool pool,
String appserver,
boolean multiSession,
boolean sessionFree)
{
this(pool, appserver, multiSession, sessionFree, null);
}
/**
* Initialize this worker.
*
* @param pool
* The service pool.
* @param appserver
* The appserver name.
* @param multiSession
* Flag indicating if it's a multi-session agent appserver.
* @param sessionFree
* Flag indicating if the appserver is Session-free.
* @param forWebServiceType
* The type of web service the worker is used for.
*/
public LegacyServiceWorker(AppServerConnectionPool pool,
String appserver,
boolean multiSession,
boolean sessionFree,
LegacyServiceHandler.ServiceType forWebServiceType)
{
this.appserver = appserver;
this.pool = pool;
this.multiSession = multiSession;
this.sessionFree = sessionFree;
this.forWebServiceType = forWebServiceType;
}
/**
* Register this procedure in the specified session, as it was created by this worker's connection.
*
* @param sessionID
* The client session which created the proxy remote procedure.
* @param proc
* The proxy remote procedure.
*/
public void addProcedure(long sessionID, ProxyProcedureWrapper proc)
{
remoteProcedures.computeIfAbsent(sessionID, (k) -> new HashSet<>());
remoteProcedures.get(sessionID).add(proc);
}
/**
* Remove the specified procedure from the given session.
*
* @param sessionID
* The client session which created the proxy remote procedure.
* @param proc
* The proxy remote procedure.
*/
public void removeProcedure(long sessionID, ProxyProcedureWrapper proc)
{
remoteProcedures.getOrDefault(sessionID, Collections.emptySet()).remove(proc);
}
/**
* Get all the proxy remote procedures created by the given session.
*
* @param sessionID
* The client session which created the proxy remote procedure.
*
* @return The set of proxy procedures still registered in {@link #remoteProcedures}.
*/
public Set<ProxyProcedureWrapper> getProcedures(long sessionID)
{
Set<ProxyProcedureWrapper> procedures = remoteProcedures
.getOrDefault(sessionID, Collections.emptySet());
return new HashSet<>(procedures);
}
/**
* Remove the remote procedures for the given session ID.
*
* @param sessionID
*/
public void removeProcedures(long sessionID)
{
remoteProcedures.remove(sessionID);
}
/**
* Get the {@link #multiSession} flag.
*
* @return See above.
*/
public boolean isMultiSession()
{
return multiSession;
}
/**
* Get the {@link AppServerHelper} for this worker.
*
* @return See above.
*/
public AppServerHelper getHelper()
{
if (helper == null || !helper.isConnected())
{
// establish the connection
helper = AppServerHelper.connect(SessionManager.get().config(),
null,
appserver,
sessionFree,
forWebServiceType == null ? null : forWebServiceType.toString());
if (helper.getConnectError() != null)
{
throw new RuntimeException(helper.getConnectError());
}
if (!pool.isWebService())
{
helper.disableLegacyErrors();
}
pool.registerWorker(helper.getConnectionID(), this);
}
return helper;
}
/**
* Post a new work to the {@link #tasks queue}.
*
* @param work
* The work to execute.
*/
public synchronized void addWork(Consumer<LegacyServiceWorker> work)
{
tasks.add(work);
this.notifyAll();
}
/**
* Notify that a task is available in the {@link #pool}.
*/
public synchronized void notifyTask()
{
this.notifyAll();
}
/**
* Stop the worker thread.
*/
public synchronized void stop()
{
this.running = false;
this.notifyAll();
}
/**
* Main processing loop - waits for task to be posted.
*/
@Override
public void run()
{
threadName = Thread.currentThread().getName();
running = true;
while (running)
{
Consumer<LegacyServiceWorker> task = null;
synchronized (this)
{
// we need to lock when checking both the pool's tasks and the worker's specific tasks; otherwise,
// if the wait and the task check are separated, race conditions can happen where the pool
// receives a task and the notifications is done before the wait can start.
task = pool.getTask();
if (task == null && !tasks.isEmpty())
{
task = tasks.removeFirst();
}
if (task == null)
{
// no task available either in the pool or specific to this worker - wait for a notification
// either from the worker's queue or from the pool's queue that something has posted
try
{
this.wait();
}
catch (InterruptedException e)
{
// ignore
}
// go to the beginning of the loop to get some task, either from the pool or from the tasks.
continue;
}
}
if (running && task != null)
{
try
{
TASK_COUNTER.update(threadName, 1);
task.accept(this);
}
catch (Throwable t)
{
LOG.logp(Level.SEVERE,
"LegacyServiceWorker",
"run",
"Problem executing the given task!",
t);
}
}
}
}
}