package com.atlassian.vcache.internal.harness;
   
   import com.atlassian.vcache.DirectExternalCache;
   import com.atlassian.vcache.LocalCacheOperations;
   import com.atlassian.vcache.PutPolicy;
   import com.atlassian.vcache.internal.RequestMetrics;
   import com.codahale.metrics.Timer;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
  
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Random;
  import java.util.stream.IntStream;
  
  import static java.util.Objects.requireNonNull;
  import static java.util.stream.Collectors.joining;
  
  public class Worker implements Runnable
  {
      private static final Logger log = LoggerFactory.getLogger(Worker.class);
      private static final String BIG_STRING =
              IntStream.range(0, 100).mapToObj(Integer::toString).collect(joining("-"));
  
      private final int id;
      private final OverallConfig overallConfig;
      private final Random numGenerator;
      private final List<String> cacheNames;
  
      private volatile boolean stopNow;
  
      public Worker(int id, OverallConfig overallConfig)
      {
          this.id = id;
          this.overallConfig = requireNonNull(overallConfig);
          this.numGenerator = new Random(id);
          this.cacheNames = new ArrayList<>(overallConfig.cacheTestConfigMap.keySet());
      }
  
      @Override
      public void run()
      {
          // The main logic is to loop until told to stop.
          // For each loop, setup the thread, perform cache operations, get the request metrics and clear the thread.
          log.info("Thread {}: running", id);
          while (!stopNow)
          {
              log.trace("Thread {}: starting loop", id);
              overallConfig.requestContextSupplier.initThread("server-mode");
  
              final int numCachesOps = numGenerator.nextInt(overallConfig.numOfCachesOps);
              log.trace("Going to perform: {} ", numCachesOps);
              IntStream.range(0, numCachesOps).forEach(i -> {
                  // Set a cache to operate on
                  final int cacheIndex = numGenerator.nextInt(cacheNames.size());
                  final String cacheName = cacheNames.get(cacheIndex);
                  log.trace("Going to operate on {}", cacheName);
                  final CacheConfig cacheConfig = overallConfig.cacheTestConfigMap.get(cacheName);
                  if (cacheConfig.cache instanceof LocalCacheOperations)
                  {
                      handleLocalCacheOperations(cacheConfig);
                  }
                  else if (cacheConfig.cache instanceof DirectExternalCache)
                  {
                      handleDirectExternalCache(cacheConfig);
                  }
                  else
                  {
                      log.warn("Ignoring unknown cache type: {}", cacheConfig.cache.getClass().getName());
                  }
              });
  
              final RequestMetrics requestMetrics =
                      overallConfig.service.metrics(overallConfig.requestContextSupplier.get());
              overallConfig.requestContextSupplier.clearThread();
              log.trace("Thread {}: stopping loop", id);
  
              // Sleep until ready to go again
              try
              {
                  Thread.sleep(numGenerator.nextInt(overallConfig.maxSleepBetweenLoops));
              }
              catch (InterruptedException e)
              {
                  Thread.interrupted();
                  throw new RuntimeException(e);
              }
          }
          log.info("Thread {}: stopping", id);
      }
  
      private void handleDirectExternalCache(CacheConfig cacheConfig)
      {
          final DirectExternalCache<String> directCache = (DirectExternalCache<String>) cacheConfig.cache;
          final int numOperations = numGenerator.nextInt(cacheConfig.maxLoops);
          final String cacheType = cacheConfig.cache.getClass().getSimpleName();
  
          IntStream.range(0, numOperations).forEach(i -> {
              final String key = "key-" + numGenerator.nextInt(cacheConfig.numOfKeys);
             randomlyPerform(
                     cacheConfig.chanceOfPutOp,
                     cacheType + ".put",
                     () -> directCache.put(key, BIG_STRING, PutPolicy.PUT_ALWAYS));
             randomlyPerform(
                     cacheConfig.chanceOfGetOp,
                     cacheType + ".get",
                     () -> directCache.get(key));
         });
     }
 
     private void handleLocalCacheOperations(CacheConfig cacheConfig)
     {
         final LocalCacheOperations<String, String> localOps = (LocalCacheOperations<String, String>) cacheConfig.cache;
         final int numOperations = numGenerator.nextInt(cacheConfig.maxLoops);
         final String cacheType = cacheConfig.cache.getClass().getSimpleName();
 
         IntStream.range(0, numOperations).forEach(i -> {
             final String key = "key-" + numGenerator.nextInt(cacheConfig.numOfKeys);
             randomlyPerform(
                     cacheConfig.chanceOfPutOp,
                     cacheType + ".put",
                     () -> localOps.put(key, BIG_STRING));
             randomlyPerform(
                     cacheConfig.chanceOfGetOp,
                     cacheType + ".get",
                     () -> localOps.get(key));
         });
     }
 
     private void randomlyPerform(int chance, String timerName, Runnable perform)
     {
         final int dice = numGenerator.nextInt(100);
         if (dice < chance)
         {
             try (Timer.Context ignored = overallConfig.metricRegistry.timer(timerName).time())
             {
                 perform.run();
             }
         }
     }
 
     void stopIt()
     {
         stopNow = true;
     }
 }