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