package com.atlassian.cache.hazelcast;
   
   import javax.annotation.Nonnull;
   
   import com.atlassian.cache.AbstractCacheLazyTest;
   import com.atlassian.cache.Cache;
   import com.atlassian.cache.CacheFactory;
   import com.atlassian.cache.CacheLoader;
   import com.atlassian.cache.CacheSettingsBuilder;
  
  import org.junit.Before;
  import org.junit.ClassRule;
  import org.junit.Test;
  
  import static org.hamcrest.MatcherAssert.assertThat;
  import static org.hamcrest.Matchers.equalTo;
  import static org.hamcrest.Matchers.is;
  import static org.hamcrest.Matchers.nullValue;
  
  public class HazelcastHybridCacheLazyTest extends AbstractCacheLazyTest
  {
      @ClassRule
      public static InitOnceHazelcastCluster cluster = InitOnceHazelcastCluster.getInstance();
  
      private Cache<String, Long> cacheNode1;
      private Cache<String, Long> cacheNode2;
  
      @Before
      public void setUp() throws Exception
      {
          cluster.reset();
          factory = HazelcastTestSupport.createDistributedFactory(cluster.getNode(0));
          CacheFactory factory2 = HazelcastTestSupport.createDistributedFactory(cluster.getNode(1));
  
          // create a loader that never creates the same value twice. Used to verify indirectly that the cache has been
          // invalidated
          CacheLoader<String, Long> loader = new BadLoader();
          // create two caches backed by the same version map, but different local caches.
          cacheNode1 = factory.getCache("test", loader, settingsBuilder().build());
          cacheNode2 = factory2.getCache("test", loader, settingsBuilder().build());
      }
  
      @Override
      protected CacheSettingsBuilder settingsBuilder()
      {
          return new CacheSettingsBuilder().remote().replicateViaInvalidation();
      }
  
      @Test
      public void testLazyCreationDoestNotInvalidateCache()
      {
          // verify that the value is cached properly
          assertThat(cacheNode1.get("key"), equalTo(1L));
          assertThat(cacheNode1.get("key"), equalTo(1L));
  
          // retrieve the same key from the other cache - should be a local cache miss and trigger a call to BadLoader
          assertThat(cacheNode2.get("key"), equalTo(2L));
  
          // but this should not trigger a cache eviction for cacheNode1
          assertThat(cacheNode1.get("key"), equalTo(1L));
      }
  
      @Test
      public void testPutInvalidatesCache()
      {
          // fill the cache on node1
          assertThat(cacheNode1.get("key"), equalTo(1L));
  
          cacheNode2.put("key", 99L);
          assertThat(cacheNode2.get("key"), equalTo(99L));
  
          // verify that cache1 has been invalidated
          assertThat(cacheNode1.get("key"), equalTo(2L));
          assertThat(cacheNode2.get("key"), equalTo(99L));
      }
  
      @Test
      public void testPutIfAbsentInvalidatesCache()
      {
          // fill the cache on node1
          assertThat(cacheNode1.get("key"), equalTo(1L));
  
          Long oldValue = cacheNode2.putIfAbsent("key", 99L);
          assertThat(oldValue, nullValue());
          assertThat(cacheNode2.get("key"), equalTo(99L));
  
          // verify that cache1 has been invalidated
          assertThat(cacheNode1.get("key"), equalTo(2L));
          assertThat(cacheNode2.get("key"), equalTo(99L));
  
          // verify that putIfAbsent does not invalidate cache when there's a value
          oldValue = cacheNode2.putIfAbsent("key", 101L);
          assertThat(oldValue, equalTo(99L));
          assertThat(cacheNode1.get("key"), equalTo(2L)); // should still be the same
      }
  
      @Test
      public void testReplaceInvalidatesCache()
      {
         // fill the caches
         assertThat(cacheNode1.get("key"), equalTo(1L));
         assertThat(cacheNode2.get("key"), equalTo(2L));
 
         cacheNode2.replace("key", 2L, 99L);
         assertThat(cacheNode1.get("key"), equalTo(3L)); // cache 1 should have invalidated
         assertThat(cacheNode2.get("key"), equalTo(99L));
 
         // verify that caches don't invalidate if the value is not replaced
         boolean replaced = cacheNode2.replace("key", 10012L, 1L);
         assertThat(replaced, is(false));
         assertThat(cacheNode1.get("key"), equalTo(3L)); // unchanged
         assertThat(cacheNode2.get("key"), equalTo(99L)); // unchanged
     }
 
     @Test
     public void testRemoveInvalidatesCache()
     {
         // fill the caches
         assertThat(cacheNode1.get("key"), equalTo(1L));
         assertThat(cacheNode2.get("key"), equalTo(2L));
 
         cacheNode2.remove("key");
         assertThat(cacheNode1.get("key"), equalTo(3L)); // invalidated
         assertThat(cacheNode2.get("key"), equalTo(4L)); // invalidated
 
         // test conditional remove
         boolean removed = cacheNode2.remove("key", 3L);
         assertThat(removed, is(false));
         assertThat(cacheNode1.get("key"), equalTo(3L)); // unchanged
         assertThat(cacheNode2.get("key"), equalTo(4L)); // unchanged
 
         removed = cacheNode2.remove("key", 4L);
         assertThat(removed, is(true));
         assertThat(cacheNode1.get("key"), equalTo(5L)); // invalidated
         assertThat(cacheNode2.get("key"), equalTo(6L)); // invalidated
     }
 
     @Test
     public void testRemoveAllInvalidatesCache()
     {
         // fill the caches
         assertThat(cacheNode1.get("key"), equalTo(1L));
         assertThat(cacheNode2.get("key"), equalTo(2L));
 
         cacheNode2.removeAll();
         assertThat(cacheNode1.get("key"), equalTo(3L)); // invalidated
         assertThat(cacheNode2.get("key"), equalTo(4L)); // invalidated
     }
 
     private static class BadLoader implements CacheLoader<String, Long>
     {
         private long value;
 
         @Nonnull
         @Override
         public Long load(@Nonnull String key)
         {
             return ++value;
         }
     }
 }