package com.atlassian.cache.hazelcast;
   
   import java.io.Serializable;
   import java.util.Map;
   
   import javax.annotation.Nonnull;
   import javax.annotation.Nullable;
   import javax.annotation.PostConstruct;
   import javax.annotation.PreDestroy;
  
  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.EntryAdapter;
  import com.hazelcast.core.EntryEvent;
  import com.hazelcast.core.HazelcastInstance;
  import com.hazelcast.core.IMap;
  import com.hazelcast.core.MembershipAdapter;
  import com.hazelcast.core.MembershipEvent;
  import com.hazelcast.map.impl.MapContainer;
  import com.hazelcast.map.impl.MapService;
  import com.hazelcast.map.impl.MapServiceContext;
  import com.hazelcast.map.impl.proxy.MapProxyImpl;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import static com.google.common.base.Preconditions.checkNotNull;
  
  /**
   * Hazelcast implementation of the {@link com.atlassian.cache.CacheManager} contract
   */
  public class HazelcastCacheManager extends AbstractCacheManager
  {
      private static final Logger log = LoggerFactory.getLogger(HazelcastCacheManager.class);
  
      protected static final String PREFIX = "atlassian-cache.";
      protected static final String PREFIX_CACHE = PREFIX + "Cache.";
      protected static final String PREFIX_CACHE_REFERENCE = PREFIX + "CacheReference.";
      protected static final String SETTINGS_MAP_NAME = PREFIX + "settings";
  
      private final HazelcastInstance hazelcast;
      private final CacheFactory localCacheFactory;
      private final IMap<String, CacheSettings> mapSettings;
      private final String mapSettingsListenerId;
      private final String membershipListenerId;
  
      private MapServiceContext mapServiceContext;
  
      public HazelcastCacheManager(HazelcastInstance hazelcast, CacheFactory localCacheFactory,
              CacheSettingsDefaultsProvider cacheSettingsDefaultsProvider)
      {
          super(cacheSettingsDefaultsProvider);
  
          this.hazelcast = hazelcast;
          this.localCacheFactory = localCacheFactory;
          this.mapSettings = hazelcast.getMap(SETTINGS_MAP_NAME);
  
          // add a listener to be notified of remote changes to the map config so the local config can be updated too
          this.mapSettingsListenerId = mapSettings.addEntryListener(new EntryAdapter<String, CacheSettings>()
          {
              @Override
              public void entryAdded(EntryEvent<String, CacheSettings> event)
              {
                  configureMap(event.getKey(), event.getValue());
              }
  
              @Override
              public void entryUpdated(EntryEvent<String, CacheSettings> event)
              {
                  reconfigureMap(event.getKey(), event.getValue());
              }
          }, true);
  
          // add a listener for join events to ensure that the local cache configs are updated
          this.membershipListenerId = hazelcast.getCluster().addMembershipListener(new MembershipAdapter()
          {
              @Override
              public void memberAdded(MembershipEvent membershipEvent)
              {
                  maybeUpdateMapContainers();
              }
          });
      }
  
     /**
      * Initializes the manager. This method must be called after the bean has been created.
      */
     @PostConstruct
     public void init()
     {
         maybeUpdateMapContainers();
     }
 
     @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();
             }
         });
     }
 
     /**
      * De-registers listeners. This method must be called when the bean is no longer required.
      */
     @PreDestroy
     public void destroy()
     {
         mapSettings.removeEntryListener(mapSettingsListenerId);
         hazelcast.getCluster().removeMembershipListener(membershipListenerId);
     }
 
     @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.findMapConfig(mapName);
         if (!mapConfig.getName().equals(mapName))
         {
             mapConfig = convertAndStoreMapConfig(mapName, settings, config, mapConfig);
 
             // store the config in the settings map to trigger other nodes to also configure the IMap
             mapSettings.putIfAbsent(mapName, asSerializable(settings));
         }
         else
         {
             log.debug("Using existing cache configuration for cache {}", mapName);
         }
 
         updateMapContainerIfAbsent(mapName, mapConfig);
 
         return mapConfig.getAsReadOnly();
     }
 
     protected MapConfig reconfigureMap(String mapName, CacheSettings newSettings)
     {
         Config config = hazelcast.getConfig();
 
         MapConfig baseConfig = config.getMapConfig(mapName);
 
         return convertAndStoreMapConfig(mapName, newSettings, config, baseConfig);
     }
 
     /**
      * Update the cache settings of an existing cache. Called by {@link com.atlassian.cache.ManagedCache} methods
      *
      * @param mapName The name of the map to update
      * @param newSettings CacheSettings object representing the new settings
      * @return true if the update succeeded, false otherwise
      */
     public boolean updateCacheSettings(@Nonnull String mapName, @Nonnull CacheSettings newSettings)
     {
         //Do nothing if the map doesn't exist
         MapConfig mapConfig = hazelcast.getConfig().findMapConfig(mapName);
         if (!mapConfig.getName().equals(mapName))
         {
             return false;
         }
 
         MapConfig newConfig = reconfigureMap(mapName, newSettings);
 
         boolean result = updateMapContainer(mapName, newConfig);
         if (result)
         {
             // store the config in the settings map to trigger other nodes to also configure the IMap
             mapSettings.put(mapName, asSerializable(newSettings));
         }
 
         return result;
     }
 
     private MapConfig convertAndStoreMapConfig(String mapName, CacheSettings newSettings, Config config, MapConfig baseConfig)
     {
         MapConfig newConfig = new MapConfig(baseConfig);
         newConfig.setName(mapName);
         newConfig.setStatisticsEnabled(true);
         new HazelcastMapConfigConfigurator().configureMapConfig(newSettings, newConfig);
         config.addMapConfig(newConfig);
         return newConfig;
     }
 
     protected <K, V> Cache<K, V> createDistributedCache(String name, CacheLoader<K, V> loader, CacheSettings settings)
     {
         final String mapName = PREFIX_CACHE + name;
         configureMap(mapName, settings);
         final IMap<K, OsgiSafe<V>> map = hazelcast.getMap(mapName);
         final CacheVersion cacheVersion = new CacheVersion(hazelcast, mapName);
 
         return new HazelcastCache<K, V>(name, map, loader, cacheVersion, this);
     }
 
     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());
 
         final String mapName = PREFIX_CACHE_REFERENCE + name;
         configureMap(mapName, overriddenSettings);
         final IMap<String, OsgiSafe<V>> map = hazelcast.getMap(mapName);
 
         return new HazelcastCachedReference<V>(name, map, supplier, this);
     }
 
     protected <K, V> Cache<K, V> createHybridCache(String name, CacheLoader<K, V> loader, CacheSettings settings)
     {
         final String mapName = PREFIX_CACHE + name;
         configureMap(mapName, settings);
         final IMap<K, Long> map = hazelcast.getMap(mapName);
 
         return new HazelcastHybridCache<K, V>(name, localCacheFactory, map, loader, this);
     }
 
     protected <V> CachedReference<V> createHybridCachedReference(String name, Supplier<V> supplier, CacheSettings settings)
     {
         final String mapName = PREFIX_CACHE_REFERENCE + name;
         configureMap(mapName, settings);
         final IMap<ReferenceKey, Long> map = hazelcast.getMap(mapName);
 
         return new HazelcastHybridCachedReference<V>(name, localCacheFactory, map, supplier, this);
     }
 
     private CacheSettings asSerializable(CacheSettings settings)
     {
         if (settings instanceof Serializable)
         {
             return settings;
         }
 
         // DefaultCacheSettings implements Serializable, make a copy
         return new CacheSettingsBuilder(settings).build();
     }
 
     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);
     }
 
     private <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);
     }
 
     private MapContainer getMapContainer(@Nonnull String name)
     {
         if (mapServiceContext == null)
         {
             MapProxyImpl<String, CacheSettings> proxy = hazelcast.getDistributedObject(MapService.SERVICE_NAME, SETTINGS_MAP_NAME);
             mapServiceContext = proxy.getService().getMapServiceContext();
         }
 
         return mapServiceContext.getMapContainer(name);
     }
 
     private boolean updateMapContainer(String mapName, MapConfig config)
     {
         final MapContainer container = getMapContainer(mapName);
         if (container == null)
         {
             return false;
         }
 
         container.setMapConfig(config);
 
         return true;
     }
 
     private boolean updateMapContainerIfAbsent(String mapName, MapConfig config)
     {
         final MapContainer container = getMapContainer(mapName);
         if (container == null || container.getMapConfig().getName().equals(config.getName()))
         {
             // container does not exist or is already configured
             return false;
         }
 
         container.setMapConfig(config);
 
         return true;
     }
 
     private void maybeUpdateMapContainers()
     {
         for (Map.Entry<String, CacheSettings> entry : mapSettings.entrySet())
         {
             configureMap(entry.getKey(), entry.getValue());
         }
     }
 
     /**
      * Get the current MapConfig for the given IMap
      *
      * @param mapName the name of the map
      * @return The current mapConfig for that map if it exists, null otherwise
      */
     @Nullable
     MapConfig getMapConfig(@Nonnull String mapName)
     {
         MapConfig mapConfig = hazelcast.getConfig().findMapConfig(mapName);
         if (!mapConfig.getName().equals(mapName))
         {
             return null;
         }
         else
         {
             return mapConfig;
         }
     }
 
     /**
      * Gets the original CacheSettings object used to configure the given map, before it was transformed into a
      * Hazelcast MapConfig
      *
      * @param mapName the name of the map
      * @return The original CacheSettings object, or null if the map doesn't exist
      */
     @Nullable
     CacheSettings getCacheSettings(@Nonnull String mapName)
     {
         return mapSettings.get(mapName);
     }
 }