package com.atlassian.cache.hazelcast;
   
   import javax.annotation.Nonnull;
   
   import com.atlassian.cache.Cache;
   import com.atlassian.cache.CacheFactory;
   import com.atlassian.cache.CacheLoader;
   import com.atlassian.cache.CacheSettings;
   import com.atlassian.cache.CacheSettingsBuilder;
  import com.atlassian.cache.CacheSettingsDefaultsProvider;
  import com.atlassian.cache.CachedReference;
  import com.atlassian.cache.ManagedCache;
  import com.atlassian.cache.Supplier;
  import com.atlassian.cache.impl.AbstractCacheManager;
  import com.atlassian.cache.impl.ReferenceKey;
  import com.atlassian.cache.impl.StrongSupplier;
  import com.atlassian.cache.impl.WeakSupplier;
  import com.atlassian.hazelcast.serialization.OsgiSafe;
  
  import com.hazelcast.config.Config;
  import com.hazelcast.config.MapConfig;
  import com.hazelcast.core.HazelcastInstance;
  import com.hazelcast.core.IMap;
  
  import static com.google.common.base.Preconditions.checkNotNull;
  
  /**
   * Hazelcast implementation of the {@link com.atlassian.cache.CacheManager} contract
   */
  public class HazelcastCacheManager extends AbstractCacheManager
  {
  
      protected static final String PREFIX = "atlassian-cache.";
      protected static final String PREFIX_CACHE = PREFIX + "Cache.";
      protected static final String PREFIX_CACHE_REFERENCE = PREFIX + "CacheReference.";
  
      private final HazelcastInstance hazelcast;
      private final CacheFactory localCacheFactory;
  
      public HazelcastCacheManager(HazelcastInstance hazelcast, CacheFactory localCacheFactory,
              CacheSettingsDefaultsProvider cacheSettingsDefaultsProvider)
      {
          super(cacheSettingsDefaultsProvider);
  
          this.hazelcast = hazelcast;
          this.localCacheFactory = localCacheFactory;
      }
  
      @Override
      protected <K, V> ManagedCache createComputingCache(@Nonnull final String name, @Nonnull final CacheSettings settings, final CacheLoader<K, V> loader)
      {
          checkSettingsAreCompatible(name, settings);
  
          // when a loader is provided, always create a new ManagedCache to ensure the correct loader is being used.
          // if the cache already existed, the backing values will be reused
          return cacheCreationLocks.get(name).withLock(new com.atlassian.util.concurrent.Supplier<ManagedCache>()
          {
              @Override
              public ManagedCache get()
              {
                  if (!caches.containsKey(name) || loader != null)
                  {
                      caches.put(name, new WeakSupplier<ManagedCache>((ManagedCache) doCreateCache(name, loader, settings)));
                  }
                  return caches.get(name).get();
              }
          });
      }
  
      @Override
      protected ManagedCache createSimpleCache(@Nonnull final String name, @Nonnull final CacheSettings settings)
      {
          checkSettingsAreCompatible(name, settings);
  
          ManagedCache existing = getManagedCache(name);
          if (existing != null)
          {
              return existing;
          }
  
          return cacheCreationLocks.get(name).withLock(new com.atlassian.util.concurrent.Supplier<ManagedCache>()
          {
              @Override
              public ManagedCache get()
              {
                  if (!caches.containsKey(name))
                  {
                      caches.put(name, new StrongSupplier<ManagedCache>((ManagedCache) doCreateCache(name, null, settings)));
                  }
                  return caches.get(name).get();
              }
          });
      }
  
      @Nonnull
      @Override
      public <V> CachedReference<V> getCachedReference(@Nonnull final String name, @Nonnull final Supplier<V> supplier,
              @Nonnull final CacheSettings settings)
      {
         checkSettingsAreCompatible(name, settings);
 
         return cacheCreationLocks.get(name).withLock(new com.atlassian.util.concurrent.Supplier<CachedReference<V>>()
         {
             @Override
             public CachedReference<V> get()
             {
                 caches.put(name, new WeakSupplier<ManagedCache>((ManagedCache) doCreateCachedReference(name, supplier, settings)));
                 //noinspection unchecked
                 return (CachedReference<V>) caches.get(name).get();
             }
         });
     }
 
     protected void checkSettingsAreCompatible(String name, CacheSettings settings)
     {
     }
 
     protected MapConfig configureMap(String mapName, CacheSettings settings)
     {
         Config config = hazelcast.getConfig();
         MapConfig mapConfig = config.getMapConfig(mapName);
         mapConfig.setStatisticsEnabled(true);
 
         new HazelcastMapConfigConfigurator().configureMapConfig(settings, mapConfig);
         config.addMapConfig(mapConfig);
 
         return mapConfig;
     }
 
     protected <K, V> Cache<K, V> createDistributedCache(String name, CacheLoader<K, V> loader, CacheSettings settings)
     {
         String mapName = PREFIX_CACHE + name;
         MapConfig config = configureMap(mapName, settings);
         IMap<K, OsgiSafe<V>> map = hazelcast.getMap(mapName);
 
         return new HazelcastCache<K, V>(name, map, config, loader, settings);
     }
 
     protected <V> CachedReference<V> createDistributedCachedReference(String name, Supplier<V> supplier, CacheSettings settings)
     {
         // override the settings to ensure the reference is flushable and the max is set to 1000. A low value for
         // maxEntries would trigger continuous cache invalidations because of the way map eviction works in Hazelcast.
         final CacheSettings overriddenSettings = checkNotNull(settings, "settings").override(
                 new CacheSettingsBuilder().flushable().maxEntries(1000).build());
 
         String mapName = PREFIX_CACHE_REFERENCE + name;
         MapConfig config = configureMap(mapName, overriddenSettings);
         IMap<String, OsgiSafe<V>> map = hazelcast.getMap(mapName);
 
         return new HazelcastCachedReference<V>(name, map, config, supplier, overriddenSettings);
     }
 
     protected <K, V> Cache<K, V> createHybridCache(String name, CacheLoader<K, V> loader, CacheSettings settings)
     {
         String mapName = PREFIX_CACHE + name;
         MapConfig config = configureMap(mapName, settings);
         IMap<K, Long> map = hazelcast.getMap(mapName);
 
         return new HazelcastHybridCache<K, V>(name, localCacheFactory, map, config, loader, settings);
     }
 
     protected <V> CachedReference<V> createHybridCachedReference(String name, Supplier<V> supplier, CacheSettings settings)
     {
         String mapName = PREFIX_CACHE_REFERENCE + name;
         MapConfig config = configureMap(mapName, settings);
         IMap<ReferenceKey, Long> map = hazelcast.getMap(mapName);
 
         return new HazelcastHybridCachedReference<V>(name, localCacheFactory, map, supplier, settings);
     }
 
     private <K, V> Cache<K, V> doCreateCache(String name, CacheLoader<K, V> loader, CacheSettings settings)
     {
         if (settings.getLocal(false))
         {
             return localCacheFactory.getCache(name, loader, settings);
         }
 
         if (settings.getReplicateViaCopy(true))
         {
             return createDistributedCache(name, loader, settings);
         }
 
         return createHybridCache(name, loader, settings);
     }
 
     protected <V> CachedReference<V> doCreateCachedReference(String name, Supplier<V> supplier, CacheSettings settings)
     {
         if (settings.getLocal(false))
         {
             return localCacheFactory.getCachedReference(name, supplier, settings);
         }
 
         // remote cached reference
         if (settings.getReplicateViaCopy(true))
         {
             return createDistributedCachedReference(name, supplier, settings);
         }
 
         return createHybridCachedReference(name, supplier, settings);
     }
 }