package com.atlassian.cache.hazelcast;
   
   import java.util.concurrent.TimeUnit;
   
   import com.atlassian.cache.Cache;
   import com.atlassian.cache.CacheSettings;
   import com.atlassian.cache.CacheSettingsBuilder;
   import com.atlassian.cache.ManagedCache;
   
  import com.hazelcast.config.InMemoryFormat;
  import com.hazelcast.config.MapConfig;
  import com.hazelcast.config.NearCacheConfig;
  import com.hazelcast.core.HazelcastInstance;
  
  import org.junit.Before;
  import org.junit.ClassRule;
  import org.junit.Test;
  
  import static org.hamcrest.MatcherAssert.assertThat;
  import static org.hamcrest.Matchers.contains;
  import static org.hamcrest.Matchers.equalTo;
  import static org.hamcrest.Matchers.hasSize;
  import static org.hamcrest.Matchers.is;
  import static org.hamcrest.Matchers.not;
  import static org.hamcrest.Matchers.notNullValue;
  import static org.hamcrest.Matchers.nullValue;
  
  public class HazelcastCacheManagerTest
  {
      @ClassRule
      public static InitOnceHazelcastCluster cluster = InitOnceHazelcastCluster.getInstance();
  
      @Before
      public void setup()
      {
          cluster.reset();
      }
  
      @Test
      public void testDefaultCacheConfigIsUsedForNewCaches()
      {
          HazelcastCacheManager factory = HazelcastTestSupport.createDistributedFactory(cluster.getNode(0), new CacheSettingsBuilder()
                  .maxEntries(10000)
                  .expireAfterAccess(6, TimeUnit.MINUTES)
                  .build());
  
          String cacheName = "defaults-test";
          Cache<String, String> cache = factory.getCache(cacheName);
  
          assertThat(cache, not(nullValue()));
          assertThat(cache.getName(), is(cacheName));
          assertThat(cache.getKeys(), hasSize(0));
  
          String mapName = HazelcastCacheManager.PREFIX_CACHE + cacheName;
  
          CacheSettings config = factory.getCacheSettings(mapName);
          assertThat(config.getMaxEntries(), is(10000));
          assertThat((int) Math.ceil(config.getExpireAfterAccess() / 1000d), is(360));
  
          factory.destroy();
      }
  
      @Test
      public void testNonFlushableCacheIsNotFlushed()
      {
          HazelcastCacheManager factory = HazelcastTestSupport.createDistributedFactory(cluster.getNode(0));
  
          // Create the "broken-toilet" cache
          Cache<String, String> nonFlushableCache = factory.getCache("broken-toilet", null, new CacheSettingsBuilder().unflushable().build());
          nonFlushableCache.put("key1", "value1");
  
          Cache<String, String> normalCache = factory.getCache("normal-cache");
          normalCache.put("key2", "value2");
  
          factory.flushCaches();
  
          nonFlushableCache = factory.getCache("broken-toilet");
          assertThat(nonFlushableCache.get("key1"), is("value1"));
  
          normalCache = factory.getCache("normal-cache");
          assertThat(normalCache.getKeys(), hasSize(0));
  
          factory.destroy();
      }
  
      @Test
      public void testTheSameCacheInstanceIsReturnedForTheSameName()
      {
          HazelcastCacheManager factory = HazelcastTestSupport.createDistributedFactory(cluster.getNode(0));
  
          final String cacheName = "same-instance-test";
          Cache<String, String> cache = factory.getCache(cacheName);
  
          cache.put("fav", "dog");
          assertThat(cache.getKeys(), contains("fav"));
          assertThat(cache.get("fav"), is("dog"));
  
          cache = factory.getCache(cacheName);
          assertThat(cache.getKeys(), contains("fav"));
         assertThat(cache.get("fav"), is("dog"));
 
         factory.destroy();
     }
 
     @Test
     public void testWildcardConfigIsUsed()
     {
         HazelcastInstance node = cluster.getNode(0);
 
         // set up a wildcard config on each of the nodes
         MapConfig wildcardConfig = node.getConfig().getMapConfig(HazelcastCacheManager.PREFIX + "*").setAsyncBackupCount(4);
         wildcardConfig.setNearCacheConfig(new NearCacheConfig()
                 .setCacheLocalEntries(true)
                 .setInMemoryFormat(InMemoryFormat.OBJECT));
         
         for (HazelcastInstance hazelcast : cluster.getNodes()) {
             hazelcast.getConfig().addMapConfig(wildcardConfig);
         }
 
         HazelcastCacheManager factory = HazelcastTestSupport.createDistributedFactory(node);
         factory.getCache("testing", null, new CacheSettingsBuilder().remote().replicateViaCopy().build());
 
         // verify that a specific MapConfig has been created for this cache _and_ that the wildcard settings have been incorporated
         MapConfig cacheConfig = node.getConfig().findMapConfig(HazelcastCacheManager.PREFIX_CACHE + "testing");
 
         assertThat(cacheConfig, notNullValue());
         assertThat(cacheConfig.getName(), equalTo(HazelcastCacheManager.PREFIX_CACHE + "testing"));
         assertThat(cacheConfig.getAsyncBackupCount(), equalTo(4));
         
         // verify that the nearCacheConfig settings from the wildcard also apply
         NearCacheConfig nearCacheConfig = cacheConfig.getNearCacheConfig();
         assertThat(nearCacheConfig, notNullValue());
         assertThat(nearCacheConfig.getInMemoryFormat(), equalTo(InMemoryFormat.OBJECT));
         assertThat(nearCacheConfig.isCacheLocalEntries(), equalTo(true));
     }
 
     @Test
     public void testUpdateCacheSettings()
     {
         HazelcastInstance node = cluster.getNode(0);
         final String cacheName = "update-cache.myCache";
 
         HazelcastCacheManager factory = HazelcastTestSupport.createDistributedFactory(node);
 
         CacheSettings cacheSettings = new CacheSettingsBuilder()
                 .maxEntries(3000)
                 .remote()
                 .flushable()
                 .expireAfterAccess(3000, TimeUnit.MILLISECONDS)
                 .expireAfterWrite(6000, TimeUnit.MILLISECONDS)
                 .replicateSynchronously()
                 .replicateViaCopy()
                 .build();
 
         //Initialize the cache
         factory.getCache(cacheName, null, cacheSettings);
 
         ManagedCache managedCache = factory.getManagedCache(cacheName);
 
         assert managedCache != null;
         assertThat(3000, equalTo(managedCache.currentMaxEntries()));
         assertThat(3000L, equalTo(managedCache.currentExpireAfterAccessMillis()));
         assertThat(6000L, equalTo(managedCache.currentExpireAfterWriteMillis()));
 
         managedCache.updateMaxEntries(5000);
         assertThat(5000, equalTo(managedCache.currentMaxEntries()));
 
         managedCache.updateExpireAfterAccess(2000L, TimeUnit.MILLISECONDS);
         assertThat(2000L, equalTo(managedCache.currentExpireAfterAccessMillis()));
 
         managedCache.updateExpireAfterWrite(8000L, TimeUnit.MILLISECONDS);
         assertThat(8000L, equalTo(managedCache.currentExpireAfterWriteMillis()));
     }
 
     @Test
     public void testNearCachesExpireEntriesBeforeIMap() throws InterruptedException
     {
         HazelcastInstance node = cluster.getNode(0);
         final String cacheName = "near-cache-expiry.testMap";
 
 
         MapConfig mapConfig = node.getConfig().getMapConfig(HazelcastCacheManager.PREFIX + cacheName);
         mapConfig.setNearCacheConfig(new NearCacheConfig()
                 .setCacheLocalEntries(true)
                 .setInMemoryFormat(InMemoryFormat.OBJECT));
 
         for (HazelcastInstance hazelcast : cluster.getNodes()) {
             hazelcast.getConfig().addMapConfig(mapConfig);
         }
 
         HazelcastCacheManager factory = HazelcastTestSupport.createDistributedFactory(node);
 
         CacheSettings cacheSettings = new CacheSettingsBuilder()
                 .expireAfterAccess(3000L, TimeUnit.MILLISECONDS)
                 .remote()
                 .flushable()
                 .maxEntries(50)
                 .replicateAsynchronously()
                 .replicateViaCopy()
                 .build();
 
         Cache<String, String> cache = factory.getCache(cacheName, null, cacheSettings);
 
         String key = cluster.generateKeyOwnedByNode(0);
         String value = "value";
 
         //Put a value, this should start the maxIdle expiry clock ticking
         cache.put(key, value);
         assertThat(cache.get(key), equalTo(value));
 
         //Continuously hitting the near cache doesn't 'tickle' the underlying cache, so our near caches should
         //Invalidate themselves infrequently but enough so that the IMap doesn't expire entries.
         for (int count=0; count<40; count++)
         {
             Thread.sleep(100L);
             assertThat("cache entry expired even after accessing the near cache!", cache.get(key), equalTo(value));
         }
     }
 }