package com.atlassian.cache.hazelcast;
   
   import java.util.Collection;
   import java.util.concurrent.TimeUnit;
   
   import javax.annotation.Nonnull;
   
   import com.atlassian.cache.Cache;
   import com.atlassian.cache.CacheEntryEvent;
  import com.atlassian.cache.CacheEntryListener;
  import com.atlassian.cache.CacheException;
  import com.atlassian.cache.CacheFactory;
  import com.atlassian.cache.CacheLoader;
  import com.atlassian.cache.CacheSettings;
  import com.atlassian.cache.CacheSettingsBuilder;
  import com.atlassian.cache.ManagedCache;
  import com.atlassian.cache.Supplier;
  import com.atlassian.cache.impl.CacheEntryListenerSupport;
  import com.atlassian.cache.impl.CacheLoaderSupplier;
  import com.atlassian.cache.impl.ValueCacheEntryListenerSupport;
  
  import com.google.common.base.Throwables;
  import com.hazelcast.core.IMap;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import static com.google.common.base.Objects.equal;
  
  /**
   * Implementation of {@link ManagedCache} and {@link Cache} that can be used when the cached values do not implement
   * {@code Serializable} but cache invalidation must work cluster-wide.
   * <p>
   * This implementation tracks versions of entries in Hazelcast {@code IMap}, but stores the actual values in a local
   * cache. Caches entries are invalidated when the cache is modified through one of the {@code put}, {@code clear},
   * {@code remove} or {@code replace} methods. Cache entries are never invalidated when {@link #get(Object)} is called,
   * not even when a value is lazily created and added to the cache.
   * <p>
   * These semantics mean that hybrid caches should only be used as computing caches (by providing a {@link
   * com.atlassian.cache.CacheLoader}) and that values should not be manually {@link #put(Object, Object)} in the cache.
   * Doing so will lead to frequent cache evictions.
   *
   * @since 2.4.0
   */
  public class HazelcastHybridCache<K, V> extends ManagedHybridCacheSupport implements Cache<K, V>
  {
  
      private static final Logger log = LoggerFactory.getLogger(HazelcastHybridCache.class);
  
      private final Cache<K, Versioned<V>> localCache;
      private final boolean selfLoading;
      private final IMap<K, Long> versionMap;
  
      private final CacheEntryListenerSupport<K, V> listenerSupport = new ValueCacheEntryListenerSupport<K, V>()
      {
          @Override
          protected void initValue(final CacheEntryListenerSupport<K, V> actualListenerSupport)
          {
              localCache.addListener(new DelegatingCacheEntryListener<K, V>(actualListenerSupport), true);
          }
  
          @Override
          protected void initValueless(final CacheEntryListenerSupport<K, V> actualListenerSupport)
          {
              localCache.addListener(new DelegatingCacheEntryListener<K, V>(actualListenerSupport), false);
          }
      };
  
      public HazelcastHybridCache(String name, CacheFactory localCacheFactory, IMap<K, Long> versionMap,
              final CacheLoader<K, V> cacheLoader, HazelcastCacheManager cacheManager)
      {
          super(name, cacheManager);
          this.selfLoading = cacheLoader != null;
  
          CacheLoader<K, Versioned<V>> versionedCacheLoader = selfLoading ? new CacheLoader<K, Versioned<V>>()
          {
              @Nonnull
              @Override
              public Versioned<V> load(@Nonnull K key)
              {
                  return loadAndVersion(key, new CacheLoaderSupplier<K, V>(key, cacheLoader));
              }
          } : null;
  
          this.versionMap = versionMap;
          this.localCache = localCacheFactory.getCache(name, versionedCacheLoader, getCacheSettings());
      }
  
      @Override
      public void clear()
      {
          removeAll();
      }
  
      @Override
      public boolean containsKey(@Nonnull final K k)
      {
          return localCache.containsKey(k);
      }
 
     @Override
     public V get(@Nonnull K key)
     {
         return getInternal(key).getValue();
     }
 
     @Nonnull
     @Override
     public V get(@Nonnull final K key, @Nonnull final Supplier<? extends V> valueSupplier)
     {
         return getInternal(key, new Supplier<Versioned<V>>()
         {
             @Override
             public Versioned<V> get()
             {
                 return loadAndVersion(key, valueSupplier);
             }
         }).getValue();
     }
 
     @Nonnull
     @Override
     public Collection<K> getKeys()
     {
         return localCache.getKeys();
     }
 
     @Nonnull
     @Override
     public String getName()
     {
         return localCache.getName();
     }
 
     @Override
     public boolean isReplicateAsynchronously()
     {
         return false;
     }
 
     @Override
     public void put(@Nonnull K key, @Nonnull V value)
     {
         Long version = incrementVersion(key);
         localCache.put(key, new Versioned<V>(value, version));
     }
 
     /**
      * {@inheritDoc}
      * <p>
      * This implementation has a weak spot: When there is an existing value, this method must return it. However, if
      * that value is stored in a remote node, it cannot be returned. <tt>null</tt> is returned instead, which signals
      * success. In order to not break any contracts, this implementation will invalidate the value on all nodes in this
      * situation.
      */
     @Override
     public V putIfAbsent(@Nonnull K key, @Nonnull V value)
     {
         Long nextVersion = getNextVersion(key);
         Versioned<V> versioned = new Versioned<V>(value, nextVersion);
         Versioned<V> oldValue = localCache.putIfAbsent(key, versioned);
         if (oldValue == null)
         {
             // entry was missing in the local cache. Increment the version to invalidate the entry across the cluster.
             // if no concurrent change to the entry was made, this should bring the version to nextVersion which
             // matches the version in the local cache. If a concurrent change was made, the version in the versionMap
             // will not match nextVersion and the local cache entry will be invalidated on the next get.
             incrementVersion(key);
 
             return null;
         }
         return oldValue.getValue();
     }
 
     @Override
     public void remove(@Nonnull K key)
     {
         // increment the version to trigger a cluster-wide invalidation
         incrementVersion(key);
         localCache.remove(key);
     }
 
     @Override
     public boolean remove(@Nonnull K key, @Nonnull V value)
     {
         Versioned<V> currentValue = null;
         try
         {
             currentValue = getInternal(key);
         }
         catch (CacheException e)
         {
             //if the get throws a CacheException we should not care it is only if the remove throws an
             //exception that we should bubble the failure
             log.debug("Swallowing exception thrown during call to remove, when looking up cache key: " + key, e);
         }
         if (currentValue != null && equal(value, currentValue.getValue()))
         {
             if (localCache.remove(key, currentValue))
             {
                 // increment the version to trigger a cluster-wide invalidation
                 incrementVersion(key);
                 return true;
             }
         }
         return false;
     }
 
     @Override
     public void removeAll()
     {
         // increment all entry versions to trigger a cluster-wide invalidation
         versionMap.executeOnEntries(IncrementVersionEntryProcessor.getInstance());
         localCache.removeAll();
     }
 
     @Override
     public boolean replace(@Nonnull K key, @Nonnull V oldValue, @Nonnull V newValue)
     {
         Versioned<V> currentValue = getInternal(key);
         if (equal(oldValue, currentValue.getValue()))
         {
             Long nextVersion = getNextVersion(key);
             if (localCache.replace(key, currentValue, new Versioned<V>(newValue, nextVersion)))
             {
                 // entry was replaced in the local cache. Increment the version to invalidate the entry across the cluster.
                 // if no concurrent change to the entry was made, this should bring the version to nextVersion which
                 // matches the version in the local cache. If a concurrent change was made, the version in the versionMap
                 // will not match nextVersion and the local cache entry will be invalidated on the next get.
                 incrementVersion(key);
                 return true;
             }
         }
 
         return false;
     }
 
     @Override
     public boolean updateExpireAfterAccess(long expireAfter, @Nonnull TimeUnit timeUnit)
     {
         if (!super.updateExpireAfterAccess(expireAfter, timeUnit))
         {
             return false;
         }
         CacheSettings settings = new CacheSettingsBuilder(getCacheSettings())
                 .expireAfterAccess(expireAfter * HazelcastMapConfigConfigurator.HYBRID_MULTIPLIER, timeUnit)
                 .build();
         cacheManager.updateCacheSettings(getHazelcastMapName(), settings);
         return true;
     }
 
     @Override
     public boolean updateExpireAfterWrite(long expireAfter, @Nonnull TimeUnit timeUnit)
     {
         if (!super.updateExpireAfterAccess(expireAfter, timeUnit))
         {
             return false;
         }
         CacheSettings settings = new CacheSettingsBuilder(getCacheSettings())
                 .expireAfterAccess(expireAfter * HazelcastMapConfigConfigurator.HYBRID_MULTIPLIER, timeUnit)
                 .build();
         cacheManager.updateCacheSettings(getHazelcastMapName(), settings);
         return true;
     }
 
     @Override
     public boolean updateMaxEntries(int newValue)
     {
         if (!super.updateMaxEntries(newValue))
         {
             return false;
         }
         CacheSettings settings = new CacheSettingsBuilder(getCacheSettings())
                 .maxEntries(HazelcastMapConfigConfigurator.HYBRID_MULTIPLIER * newValue)
                 .build();
         cacheManager.updateCacheSettings(getHazelcastMapName(), settings);
         return true;
     }
 
     @Override
     public void addListener(@Nonnull CacheEntryListener<K, V> listener, boolean includeValues)
     {
         listenerSupport.add(listener, includeValues);
     }
 
     @Override
     public void removeListener(@Nonnull CacheEntryListener<K, V> listener)
     {
         listenerSupport.remove(listener);
     }
 
     @Override
     protected ManagedCache getLocalCache()
     {
         return (ManagedCache) localCache;
     }
 
     private CacheSettings getCacheSettings()
     {
         return cacheManager.getCacheSettings(getHazelcastMapName());
     }
 
     private String getHazelcastMapName()
     {
         return versionMap.getName();
     }
 
     @Override
     public boolean isFlushable()
     {
         return getCacheSettings().getFlushable(true);
     }
 
     private Versioned<V> loadAndVersion(final K key, Supplier<? extends V> supplier)
     {
         try
         {
             // retrieve the current version if it exists and is compatible with the generated value (identical hash),
             // otherwise generate or increment the tracked version
             long version = getVersion(key);
 
             V value = supplier.get();
             //noinspection ConstantConditions
             if (value == null)
             {
                 throw new CacheException("The generated value for cache '" + getName() + "' was null for key '" +
                         key + "'. Null values are not supported.");
             }
 
             log.debug("Generated value '{}' for key '{}' in cache with name '{}'", value, key, localCache.getName());
             return new Versioned<V>(value, version);
         }
         catch (RuntimeException e)
         {
             Throwables.propagateIfInstanceOf(e, CacheException.class);
             throw new CacheException("Error generating a value for key '" + key + "' in cache '" + localCache.getName() + "'", e);
         }
     }
 
     @Nonnull
     private Versioned<V> getInternal(K key)
     {
         Versioned<V> versioned = localCache.get(key);
         if (versioned != null)
         {
             Long version = versionMap.get(key);
             if (version != null && version == versioned.getVersion())
             {
                 // versions match, cache is up to date
                 return versioned;
             }
 
             // Value in localCache is outdated, clear it.
             localCache.remove(key);
 
             if (selfLoading)
             {
                 // a new value will be recalculated and the version will be properly initialized
                 //noinspection ConstantConditions
                 return localCache.get(key);
             }
         }
         return Versioned.empty();
     }
 
     @Nonnull
     private Versioned<V> getInternal(K key, Supplier<Versioned<V>> valueSupplier)
     {
         Versioned<V> versioned = localCache.get(key, valueSupplier);
         Long version = versionMap.get(key);
         if (version != null && version == versioned.getVersion())
         {
             // versions match, cache is up to date
             return versioned;
         }
 
         // Value in localCache is outdated, clear it.
         localCache.remove(key);
 
         // a new value will be recalculated and the version will be properly initialized
         return localCache.get(key, valueSupplier);
     }
 
     private Long getNextVersion(K key)
     {
         Long version = versionMap.get(key);
         return version == null ? 1L : version + 1L;
     }
 
     private Long getVersion(K key)
     {
         // try a standard get first to give the near-cache a chance
         Long version = versionMap.get(key);
         if (version == null)
         {
             version = (Long) versionMap.executeOnKey(key, GetOrInitVersionEntryProcessor.getInstance());
         }
         return version;
     }
 
     private Long incrementVersion(K key)
     {
         return (Long) versionMap.executeOnKey(key, IncrementVersionEntryProcessor.getInstance());
     }
 
     private static class DelegatingCacheEntryListener<K, V> implements CacheEntryListener<K, Versioned<V>>
     {
         private final CacheEntryListenerSupport<K, V> listenerSupport;
 
         private DelegatingCacheEntryListener(final CacheEntryListenerSupport<K, V> listenerSupport)
         {
             this.listenerSupport = listenerSupport;
         }
 
         @Override
         public void onAdd(@Nonnull CacheEntryEvent<K, Versioned<V>> event)
         {
             listenerSupport.notifyAdd(event.getKey(), get(event.getValue()));
         }
 
         @Override
         public void onEvict(@Nonnull CacheEntryEvent<K, Versioned<V>> event)
         {
             listenerSupport.notifyEvict(event.getKey(), get(event.getOldValue()));
         }
 
         @Override
         public void onRemove(@Nonnull CacheEntryEvent<K, Versioned<V>> event)
         {
             listenerSupport.notifyRemove(event.getKey(), get(event.getOldValue()));
         }
 
         @Override
         public void onUpdate(@Nonnull CacheEntryEvent<K, Versioned<V>> event)
         {
             listenerSupport.notifyUpdate(event.getKey(), get(event.getValue()), get(event.getOldValue()));
         }
 
         private V get(Versioned<V> versioned)
         {
             return versioned != null ? versioned.getValue() : null;
         }
     }
 }