package com.atlassian.vcache.internal.core.service;
   
   import com.atlassian.utt.concurrency.Barrier;
   import com.atlassian.vcache.RequestCache;
   import com.atlassian.vcache.internal.RequestContext;
   import com.atlassian.vcache.internal.core.DefaultRequestContext;
   import com.atlassian.vcache.internal.test.AbstractRequestCacheTest;
   import com.google.common.base.Throwables;
   import org.junit.After;
  import org.junit.Before;
  import org.junit.Test;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import javax.annotation.Nullable;
  import java.util.Map;
  import java.util.Optional;
  import java.util.concurrent.Callable;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.stream.Collectors;
  import java.util.stream.IntStream;
  
  import static org.hamcrest.Matchers.is;
  import static org.junit.Assert.assertThat;
  
  public class DefaultRequestCacheTest extends AbstractRequestCacheTest {
      private static final Logger log = LoggerFactory.getLogger(DefaultRequestCacheTest.class);
  
      private final ExecutorService executorService = Executors.newCachedThreadPool();
  
      @Nullable
      private RequestContext context;
  
      @Before
      public void initContext() {
          context = new DefaultRequestContext("tenant-1");
      }
  
      @After
      public void tearDown() {
          executorService.shutdown();
      }
  
      @Override
      protected <K, V> RequestCache<K, V> createCache(String name) {
          return new DefaultRequestCache<>(name, () -> context);
      }
  
      @Test
      public void repeated_concurrency() {
          IntStream.range(0, 1_000).forEach(i -> {
              both_getBulk_and_get_fail();
              getBulk_fail_and_get_succeed();
              getBulk_pass_and_get_reuse();
          });
      }
  
      @Test
      public void both_getBulk_and_get_fail() {
          //Scenario:
          // T1: bulkGet A, B, C
          // T2: get B, supplier
          // T1: bulkGet completes load with an exception
          // T2: get B, supplier -> fails with another exception
  
          final Barrier blockedInFactory = new Barrier();
          final Barrier resumeInFactory = new Barrier();
  
          final DefaultRequestCache<String, String> cache = new DefaultRequestCache<>("cache", () -> context);
  
          final RuntimeException exBulk = new RuntimeException("getBulk() exception");
          final CompletableFuture<Map<String, String>> t1Result = runAndWaitForStart(() -> cache.getBulk(keys -> {
              log.debug("{}: about to wait", Thread.currentThread().getName());
              blockedInFactory.signal();
              resumeInFactory.await();
              log.debug("{}: About to throw exception", Thread.currentThread().getName());
              throw exBulk;
          }, "A", "B", "C"));
  
          blockedInFactory.await();
          final RuntimeException exSingle = new RuntimeException("get() exception");
          final CompletableFuture<String> t2Result = runAndWaitForStart(() -> cache.get("B", () -> {
              log.debug("{}: About to throw exception", Thread.currentThread().getName());
              throw exSingle;
          }));
  
          log.debug("{}: About to signal on barrierBulk", Thread.currentThread().getName());
          resumeInFactory.signal(); //Wait until T2 cache.get calls join
  
          assertThat(t1Result.handle((r, e) -> Throwables.getRootCause(e)).join(), is(exBulk));
          assertThat(t2Result.handle((r, e) -> Throwables.getRootCause(e)).join(), is(exSingle));
          assertThat(cache.get("A"), is(Optional.empty()));
          assertThat(cache.get("B"), is(Optional.empty()));
          assertThat(cache.get("C"), is(Optional.empty()));
      }
 
     @Test
     public void getBulk_fail_and_get_succeed() {
         //Scenario:
         // T1: bulkGet A, B, C
         // T2: get B, supplier
         // T1: bulkGet function throws exception
         // T2: get B, supplier -> returns value
 
         final Barrier blockedInFactory = new Barrier();
         final Barrier resumeInFactory = new Barrier();
 
         final DefaultRequestCache<String, String> cache = new DefaultRequestCache<>("cache", () -> context);
 
         final RuntimeException exBulk = new RuntimeException("getBulk() exception");
         final CompletableFuture<Map<String, String>> t1Result = runAndWaitForStart(() -> cache.getBulk(keys -> {
             log.debug("{}: about to wait", Thread.currentThread().getName());
             blockedInFactory.signal();
             resumeInFactory.await();
             log.debug("{}: About to throw exception", Thread.currentThread().getName());
             throw exBulk;
         }, "A", "B", "C"));
 
         blockedInFactory.await();
         final CompletableFuture<String> t2Result = runAndWaitForStart(() -> cache.get("B", () -> "bee"));
 
         log.debug("{}: About to signal on barrierBulk", Thread.currentThread().getName());
         resumeInFactory.signal(); //Wait until T2 cache.get calls join
 
         assertThat(t1Result.handle((r, e) -> Throwables.getRootCause(e)).join(), is(exBulk));
         assertThat(t2Result.join(), is("bee"));
         assertThat(cache.get("A"), is(Optional.empty()));
         assertThat(cache.get("B"), is(Optional.of("bee")));
         assertThat(cache.get("C"), is(Optional.empty()));
     }
 
     @Test
     public void getBulk_pass_and_get_reuse() {
         //Scenario:
         // T1: bulkGet A, B, C
         // T2: get B, supplier
         // T1: bulkGet function returns values
         // T2: get B, supplier -> returns bulkGet value
 
         final Barrier blockedInFactory = new Barrier();
         final Barrier resumeInFactory = new Barrier();
 
         final DefaultRequestCache<String, String> cache = new DefaultRequestCache<>("cache", () -> context);
 
         final CompletableFuture<Map<String, String>> t1Result = runAndWaitForStart(() -> cache.getBulk(keys -> {
             log.debug("{}: About to await on barrierBulk", Thread.currentThread().getName());
             blockedInFactory.signal();
             resumeInFactory.await();
             log.debug("{}: resuming", Thread.currentThread().getName());
             return keys.stream().collect(Collectors.toMap(k -> k, k -> k + "-1"));
         }, "A", "B", "C"));
 
         blockedInFactory.await();
         final CompletableFuture<String> t2Result = runAndWaitForStart(() -> cache.get("B", () -> "bee"));
 
         log.debug("{}: About to signal on barrierBulk", Thread.currentThread().getName());
         resumeInFactory.signal(); //Wait until T2 cache.get calls join
         log.debug("{}: and now", Thread.currentThread().getName());
 
 
         final Map<String, String> t1ResultData = t1Result.join();
         assertThat(t1ResultData.get("A"), is("A-1"));
         assertThat(t1ResultData.get("B"), is("B-1"));
         assertThat(t1ResultData.get("C"), is("C-1"));
         assertThat(t2Result.join(), is("B-1"));
         assertThat(cache.get("A"), is(Optional.of("A-1")));
         assertThat(cache.get("B"), is(Optional.of("B-1")));
         assertThat(cache.get("C"), is(Optional.of("C-1")));
     }
 
     private <T> CompletableFuture<T> runAndWaitForStart(Callable<T> r) {
         final Barrier barrier = new Barrier();
         log.debug("{}: About to submit()", Thread.currentThread().getName());
         final Future<T> result = executorService.submit(() -> {
             log.debug("{}: About to signal()", Thread.currentThread().getName());
             barrier.signal();
             log.debug("{}: resume after signal", Thread.currentThread().getName());
             return r.call();
         });
 
         log.debug("{}: About to await()", Thread.currentThread().getName());
         barrier.await();
         log.debug("{}: resume after await()", Thread.currentThread().getName());
 
         return CompletableFuture.supplyAsync(() -> {
             try {
                 return result.get();
             } catch (InterruptedException | ExecutionException e) {
                 throw new RuntimeException(e);
             }
         });
     }
 }