content: /*
   * Copyright 2023 Adaptive Financial Consulting
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   * https://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package com.aeroncookbook.rfq;
  
  import com.aeroncookbook.rfq.infra.AppClusteredService;
  import io.aeron.cluster.ClusteredMediaDriver;
  import io.aeron.cluster.service.ClusteredServiceContainer;
  import io.aeron.samples.cluster.ClusterConfig;
  import org.agrona.concurrent.ShutdownSignalBarrier;
  import org.agrona.concurrent.SystemEpochClock;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.net.InetAddress;
  import java.net.UnknownHostException;
  import java.util.List;
  import java.util.concurrent.TimeUnit;
  
  import static java.lang.Integer.parseInt;
  
  /**
   * Sample cluster application
   */
  public class ClusterApp
  {
      private static final Logger LOGGER = LoggerFactory.getLogger(ClusterApp.class);
  
      /**
       * The main method.
       *
       * @param args command line args
       */
      @SuppressWarnings("try")
      public static void main(final String[] args)
      {
          final int portBase = getBasePort();
          final int nodeId = getClusterNode();
          final String hosts = getClusterAddresses();
  
          final List<String> hostAddresses = List.of(hosts.split(","));
          final ClusterConfig clusterConfig = ClusterConfig.create(nodeId, hostAddresses, hostAddresses, portBase,
              new AppClusteredService());
          clusterConfig.consensusModuleContext().ingressChannel("aeron:udp");
  
          //this may need tuning for your environment.
          clusterConfig.consensusModuleContext().leaderHeartbeatTimeoutNs(TimeUnit.SECONDS.toNanos(3));
          clusterConfig.errorHandler(Throwable::printStackTrace);
  
          awaitDnsResolution(hostAddresses, nodeId);
  
          try (
              ShutdownSignalBarrier barrier = new ShutdownSignalBarrier();
              ClusteredMediaDriver ignored = ClusteredMediaDriver.launch(
                  clusterConfig.mediaDriverContext().terminationHook(barrier::signalAll),
                  clusterConfig.archiveContext(),
                  clusterConfig.consensusModuleContext().terminationHook(barrier::signalAll));
              ClusteredServiceContainer ignored1 = ClusteredServiceContainer.launch(
                  clusterConfig.clusteredServiceContext().terminationHook(barrier::signalAll)))
          {
              LOGGER.info("Started Cluster Node...");
              barrier.await();
              LOGGER.info("Exiting");
          }
      }
  
      /**
       * Read the cluster addresses from the environment variable CLUSTER_ADDRESSES or the
       * system property cluster.addresses
       *
       * @return cluster addresses
       */
      private static String getClusterAddresses()
      {
          String clusterAddresses = System.getenv("CLUSTER_ADDRESSES");
          if (null == clusterAddresses || clusterAddresses.isEmpty())
          {
              clusterAddresses = System.getProperty("cluster.addresses", "localhost");
          }
          return clusterAddresses;
      }
  
      /**
       * Get the cluster node id
       *
       * @return cluster node id, default 0
       */
      private static int getClusterNode()
      {
          String clusterNode = System.getenv("CLUSTER_NODE");
          if (null == clusterNode || clusterNode.isEmpty())
          {
              clusterNode = System.getProperty("node.id", "0");
          }
          return parseInt(clusterNode);
      }
  
      /**
       * Get the base port for the cluster configuration
       *
       * @return base port, default 9000
       */
      private static int getBasePort()
      {
          String portBaseString = System.getenv("CLUSTER_PORT_BASE");
          if (null == portBaseString || portBaseString.isEmpty())
          {
              portBaseString = System.getProperty("port.base", "9000");
          }
          return parseInt(portBaseString);
      }
  
      /**
       * Await DNS resolution of self. Under Kubernetes, this can take a while.
       *
       * @param hostArray host array
       * @param nodeId    node id
       */
      private static void awaitDnsResolution(final List<String> hostArray, final int nodeId)
      {
          if (applyDnsDelay())
          {
              LOGGER.info("Waiting 5 seconds for DNS to be registered...");
              quietSleep(5000);
          }
  
          final long endTime = SystemEpochClock.INSTANCE.time() + 60000;
          final String nodeName = hostArray.get(nodeId);
          java.security.Security.setProperty("networkaddress.cache.ttl", "0");
  
          boolean resolved = false;
          while (!resolved)
          {
              if (SystemEpochClock.INSTANCE.time() > endTime)
              {
                  LOGGER.error("cannot resolve name {}, exiting", nodeName);
                  System.exit(-1);
              }
  
              try
              {
                  InetAddress.getByName(nodeName);
                  resolved = true;
              }
              catch (final UnknownHostException e)
              {
                  LOGGER.warn("cannot yet resolve name {}, retrying in 3 seconds", nodeName);
                  quietSleep(3000);
              }
          }
      }
  
      /**
       * Sleeps for the given number of milliseconds, ignoring any interrupts.
       *
       * @param millis the number of milliseconds to sleep.
       */
      private static void quietSleep(final long millis)
      {
          try
          {
              Thread.sleep(millis);
          }
          catch (final InterruptedException ex)
          {
              LOGGER.warn("Interrupted while sleeping");
          }
      }
  
      /**
       * Apply DNS delay
       *
       * @return true if DNS delay should be applied
       */
      private static boolean applyDnsDelay()
      {
          final String dnsDelay = System.getenv("DNS_DELAY");
          if (null == dnsDelay || dnsDelay.isEmpty())
          {
              return false;
          }
          return Boolean.parseBoolean(dnsDelay);
      }
  }
  