package com.atlassian.cache;
   
   import java.util.concurrent.Callable;
   import java.util.concurrent.CountDownLatch;
   import java.util.concurrent.ExecutionException;
   import java.util.concurrent.ExecutorService;
   import java.util.concurrent.Executors;
   import java.util.concurrent.Future;
   import java.util.concurrent.TimeUnit;
  import java.util.concurrent.TimeoutException;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.concurrent.atomic.AtomicLong;
  import javax.annotation.Nonnull;
  
  import org.hamcrest.Matchers;
  import org.junit.Test;
  
  import static org.hamcrest.Matchers.hasSize;
  import static org.hamcrest.Matchers.is;
  import static org.junit.Assert.assertThat;
  
  public abstract class AbstractCacheTest
  {
      @Test
      public void testRemoveFromCacheWhileLoadingCacheValueReturnsLatestValue()
              throws InterruptedException, TimeoutException, ExecutionException
      {
          testWhenCacheValueManipulatedNewValueReturned(new CacheManipulator<String, Long>()
          {
              @Override
              public void manipulateKeyOn(final String key, final Cache<String, Long> cache)
              {
                  cache.remove(key);
              }
          });
      }
  
      /**
       * Confirms that when a remove of a key with a specific value is called, if there is a loading call which is paused
       * we wait for that load to finish before starting the remove.
       *
       * <b>Note</b> this test will become flakey when the code covering the code path is bad.
       * @throws InterruptedException
       * @throws TimeoutException
       * @throws ExecutionException
       */
      @Test
      public void testRemoveSpecificValueFromCacheWhileLoadingCacheValueReturnsLatestValue()
              throws InterruptedException, TimeoutException, ExecutionException
      {
          testWhenCacheValueManipulatedNewValueReturned(new CacheManipulator<String, Long>()
          {
              @Override
              public void manipulateKeyOn(final String key, final Cache<String, Long> cache)
              {
                  //the initial value of the cache should be 1. In essence this test should operate in the same way that
                  //#testRemoveFromCacheWhileLoadingCacheValueReturnsLatestValue does
                  cache.remove(key, 1L);
              }
          });
      }
  
      protected CacheFactory factory;
  
      protected Cache<String,Long> makeExceptionalCache()
      {
          // Build a Cache using the builder
          CacheLoader<String, Long> loader = new CacheLoader<String, Long>()
          {
              @Nonnull
              @Override
              public Long load(@Nonnull final String key)
              {
                  return Long.valueOf(key);
              }
          };
          final Cache<String, Long> cache = factory.getCache("mycache", loader, settingsBuilder().build());
          assertEmpty(cache);
          return cache;
      }
  
      protected Cache<String, Long> makeExpiringCache()
      {
          // Build a Cache using the builder
          CacheLoader<String, Long> loader = new CacheLoader<String, Long>()
          {
              @Nonnull
              @Override
              public Long load(@Nonnull final String key)
              {
                  try
                  {
                      return Long.valueOf(key);
                  }
                  catch (NumberFormatException e)
                  {
                      return -21L;
                  }
             }
         };
         CacheSettings settings = settingsBuilder().expireAfterAccess(10, TimeUnit.MILLISECONDS).build();
         final Cache<String, Long> cache = factory.getCache("mycache", loader, settings);
         assertEmpty(cache);
         return cache;
     }
 
     protected Cache<String,Long> makeNullReturningCache()
     {
         // Build a Cache using the builder
         CacheLoader<String, Long> loader = new CacheLoader<String, Long>()
         {
             @Nonnull
             @Override
             public Long load(@Nonnull final String key)
             {
                 try
                 {
                     return Long.valueOf(key);
                 }
                 catch (NumberFormatException e)
                 {
                     return null;
                 }
             }
         };
         final Cache<String, Long> cache = factory.getCache("mycache", loader, settingsBuilder().build());
         assertEmpty(cache);
         return cache;
     }
 
     protected Cache<String, Long> makeSimpleCache()
     {
         // Build a Cache using the builder
         final Cache<String, Long> cache = factory.getCache("mycache", null, settingsBuilder().build());
         assertEmpty(cache);
         return cache;
     }
 
     protected Cache<String, Long> makeSizeLimitedCache(int maxEntries)
     {
         CacheSettings required = settingsBuilder().maxEntries(maxEntries).build();
         final Cache<String, Long> cache = factory.getCache("mycache", null, required);
         assertEmpty(cache);
         return cache;
     }
 
     protected Cache<String, Long> makeSizeLimitedCache(final int maxEntries, final AtomicInteger loadCounter)
     {
         // Build a Cache using the builder
         CacheLoader<String, Long> loader = new CacheLoader<String, Long>()
         {
             @Nonnull
             @Override
             public Long load(@Nonnull final String key)
             {
                 loadCounter.incrementAndGet();
                 return Long.valueOf(key);
             }
         };
         CacheSettings settings = settingsBuilder().maxEntries(maxEntries).build();
         final Cache<String, Long> cache = factory.getCache("mycache", loader, settings);
         assertEmpty(cache);
         return cache;
     }
 
     protected Cache<String, Long> makeUnexpiringCache()
     {
         // Build a Cache using the builder
         CacheLoader<String, Long> loader = new CacheLoader<String, Long>()
         {
             @Nonnull
             @Override
             public Long load(@Nonnull final String key)
             {
                 try
                 {
                     return Long.valueOf(key);
                 }
                 catch (NumberFormatException e)
                 {
                     return -21L;
                 }
             }
         };
         final Cache<String, Long> cache = factory.getCache("mycache", loader, settingsBuilder().build());
         assertEmpty(cache);
         return cache;
     }
 
     protected CacheSettingsBuilder settingsBuilder()
     {
         return new CacheSettingsBuilder();
     }
 
     // Hiding broken type inferences
 
     protected static <K,V> void assertEmpty(Cache<K,V> cache)
     {
         assertThat(cache.getKeys(), Matchers.<K>empty());
     }
 
     protected static <K,V> void assertSize(Cache<K,V> cache, final int expectedSize)
     {
         assertThat(cache.getKeys(), hasSize(expectedSize));
     }
 
     protected Cache<String, Long> makeCacheUsingLoader(final CacheLoader<String, Long> loader)
     {
         final Cache<String, Long> cache = factory.getCache("mycache", loader, settingsBuilder().build());
         return cache;
     }
 
     private void testWhenCacheValueManipulatedNewValueReturned(final CacheManipulator<String, Long> remover)
             throws InterruptedException, TimeoutException, ExecutionException
     {
         final CountDownLatch readyToRemove = new CountDownLatch(1);
         final CountDownLatch hasStartedRemove = new CountDownLatch(1);
         final AtomicLong cacheValue = new AtomicLong(1L);
         final AtomicBoolean loaderHasReturned = new AtomicBoolean();
         final String key = "keyvalue";
         //Vis value can cause the tests to fail in CI depending on the speed of the agent running the tests
         final CacheLoader<String, Long> loader = new CacheLoader<String, Long>()
         {
             @Nonnull
             @Override
             public Long load(@Nonnull final String key)
             {
                 try
                 {
                     //Get the initial value before signalling to the other thread that it's safe to increment the
                     //initial value
                     Long value = cacheValue.get();
                     readyToRemove.countDown();
                     //Each of the caches implements blocking in a different way
                     // Guava holds a future waiting for the loader to return
                     // Ehcache SelfPopulatingCache inherits from BlockingCache which locks on the key
                     // Hazlecast locks on the key of the value for five minutes
                     // The timeout prevents a deadlock when the call to the loader is blocked and the remove then also
                     //waits on the blocked loader
                     hasStartedRemove.await();
                     //Give enough time for the outside thread to do its manipulation of the cache
                     Thread.sleep(200l);
                     loaderHasReturned.set(true);
                     return value;
                 }
                 catch (InterruptedException e)
                 {
                     throw new RuntimeException(e);
                 }
             }
         };
 
         final Cache<String, Long> pausingCache = makeCacheUsingLoader(loader);
 
         ExecutorService service = Executors.newFixedThreadPool(2);
         Future<Long> f = service.submit(new Callable<Long>()
         {
             @Override
             public Long call() throws Exception
             {
                 return pausingCache.get(key);
             }
         });
 
         readyToRemove.await();
         Long newCacheValue = cacheValue.incrementAndGet();
 
         Future<Void> manipulationOp = service.submit(new Callable<Void>()
         {
             @Override
             public Void call() throws Exception
             {
                 //Simulate a change to the underlying value
                 hasStartedRemove.countDown();
                 remover.manipulateKeyOn(key, pausingCache);
                 int i = 0;
                 while(!loaderHasReturned.get() && i < 20)
                 {
                     //Different hardware and jvms has us get to this point at different times we wait for at most a few seconds
                     //for the loader to return then we consider whether we have failed or not
                     Thread.sleep(100L);
                 }
                 return null;
             }
         });
 
         manipulationOp.get();
 
         assertThat(loaderHasReturned.get(), is(true));
         Long initialCacheValue = f.get(400, TimeUnit.MILLISECONDS);
 
         Long cachedValue = pausingCache.get(key);
         assertThat(initialCacheValue, is(1L));
         assertThat(cachedValue, is(newCacheValue));
     }
 }