package com.atlassian.cache.hazelcast;
   
   import java.lang.ref.WeakReference;
   import java.util.Collection;
   
   import javax.annotation.Nonnull;
   
   import com.atlassian.cache.Cache;
   import com.atlassian.cache.CacheEntryEvent;
  import com.atlassian.cache.CacheEntryListener;
  import com.atlassian.cache.CacheFactory;
  import com.atlassian.cache.CacheLoader;
  import com.atlassian.cache.CacheSettings;
  import com.atlassian.cache.ManagedCache;
  import com.atlassian.cache.Supplier;
  import com.atlassian.cache.impl.CacheEntryListenerSupport;
  import com.atlassian.cache.impl.ValueCacheEntryListenerSupport;
  
  import com.hazelcast.core.Cluster;
  import com.hazelcast.core.ITopic;
  import com.hazelcast.core.MembershipAdapter;
  import com.hazelcast.core.MembershipEvent;
  import com.hazelcast.core.Message;
  import com.hazelcast.core.MessageListener;
  
  /**
   * Implementation of {@link ManagedCache} and {@link Cache} that does not require the values to be {@code Serializable}.
   * It works by caching values locally and broadcasting invalidations cluster-wide using a Hazelcast {@link ITopic}.
   * <p>
   * Cache entries are invalidated when the cache is modified through one of the methods {@link #clear}, {@link #remove},
   * {@link #replace}, or {@link #put} (when it overwrites an existing value).
   *
   * @since 2.12.0
   */
  public class HazelcastAsyncHybridCache<K, V> extends ManagedHybridCacheSupport implements Cache<K, V>
  {
      private final AsyncInvalidationListener<K> listener;
      private final Cache<K, V> localCache;
      private final CacheEntryListenerSupport<K, V> listenerSupport = new ValueCacheEntryListenerSupport<K, V>()
      {
          @Override
          protected void initValue(final CacheEntryListenerSupport<K, V> actualListenerSupport)
          {
              localCache.addListener(new DelegatingCacheEntryListener<>(actualListenerSupport), true);
          }
  
          @Override
          protected void initValueless(final CacheEntryListenerSupport<K, V> actualListenerSupport)
          {
              localCache.addListener(new DelegatingCacheEntryListener<>(actualListenerSupport), false);
          }
      };
  
      public HazelcastAsyncHybridCache(String name, CacheFactory localCacheFactory, ITopic<K> invalidationTopic,
              final CacheLoader<K, V> cacheLoader, HazelcastCacheManager cacheManager, CacheSettings settings)
      {
          super(name, cacheManager);
          localCache = localCacheFactory.getCache(name, cacheLoader, settings);
          listener = new AsyncInvalidationListener<K>(cacheManager.getHazelcastInstance().getCluster(), localCache,
                  invalidationTopic);
      }
  
      @Override
      public void clear()
      {
          removeAll();
      }
  
      @Override
      public boolean containsKey(@Nonnull final K key)
      {
          return localCache.containsKey(key);
      }
  
      @Override
      public V get(@Nonnull K key)
      {
          return localCache.get(key);
      }
  
      @Nonnull
      @Override
      public V get(@Nonnull final K key, @Nonnull final Supplier<? extends V> valueSupplier)
      {
          return localCache.get(key, valueSupplier);
      }
  
      @Nonnull
      @Override
      public Collection<K> getKeys()
      {
          return localCache.getKeys();
      }
  
      @Nonnull
      @Override
      public String getName()
      {
          return localCache.getName();
     }
 
     @Override
     public boolean isFlushable()
     {
         return true;
     }
 
     @Override
     public boolean isReplicateAsynchronously()
     {
         return true;
     }
 
     @Override
     public void put(@Nonnull K key, @Nonnull V value)
     {
         invalidateRemotely(key);
         localCache.put(key, value);
     }
 
     @Override
     public V putIfAbsent(@Nonnull K key, @Nonnull V value)
     {
         V oldValue = localCache.putIfAbsent(key, value);
         if (oldValue == null)
         {
             invalidateRemotely(key);
         }
         return oldValue;
     }
 
     @Override
     public void remove(@Nonnull K key)
     {
         invalidateRemotely(key);
         localCache.remove(key);
     }
 
     @Override
     public boolean remove(@Nonnull K key, @Nonnull V value)
     {
         if (localCache.remove(key, value))
         {
             invalidateRemotely(key);
             return true;
         }
         return false;
     }
 
     @Override
     public void removeAll()
     {
         invalidateRemotely();
         localCache.removeAll();
     }
 
     @Override
     public boolean replace(@Nonnull K key, @Nonnull V oldValue, @Nonnull V newValue)
     {
         if (localCache.replace(key, oldValue, newValue))
         {
             // entry was replaced in the local cache. Invalidate the entry across the cluster.
             invalidateRemotely(key);
             return true;
         }
 
         return false;
     }
 
     @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 void invalidateRemotely()
     {
         listener.publish(null);
     }
 
     private void invalidateRemotely(@Nonnull K key)
     {
         listener.publish(key);
     }
 
     private static class DelegatingCacheEntryListener<K, V> implements CacheEntryListener<K, V>
     {
         private final CacheEntryListenerSupport<K, V> listenerSupport;
 
         private DelegatingCacheEntryListener(final CacheEntryListenerSupport<K, V> listenerSupport)
         {
             this.listenerSupport = listenerSupport;
         }
 
         @Override
         public void onAdd(@Nonnull CacheEntryEvent<K, V> event)
         {
             listenerSupport.notifyAdd(event.getKey(), event.getValue());
         }
 
         @Override
         public void onEvict(@Nonnull CacheEntryEvent<K, V> event)
         {
             listenerSupport.notifyEvict(event.getKey(), event.getOldValue());
         }
 
         @Override
         public void onRemove(@Nonnull CacheEntryEvent<K, V> event)
         {
             listenerSupport.notifyRemove(event.getKey(), event.getOldValue());
         }
 
         @Override
         public void onUpdate(@Nonnull CacheEntryEvent<K, V> event)
         {
             listenerSupport.notifyUpdate(event.getKey(), event.getValue(), event.getOldValue());
         }
     }
 
     private static class AsyncInvalidationListener<K> extends MembershipAdapter implements MessageListener<K> {
 
         private final Cluster cluster;
         private final WeakReference<Cache<K, ?>> localCacheRef;
         private final String membershipListenerId;
         private final ITopic<K> topic;
         private final String topicListenerId;
 
         AsyncInvalidationListener(Cluster cluster, Cache<K, ?> localCache, ITopic<K> topic)
         {
             this.cluster = cluster;
             this.localCacheRef = new WeakReference<>(localCache);
             this.topic = topic;
             this.topicListenerId = topic.addMessageListener(this);
             this.membershipListenerId = cluster.addMembershipListener(this);
         }
 
         @Override
         public void memberAdded(MembershipEvent membershipEvent)
         {
             Cache<K, ?> localCache = localCacheRef.get();
             if (localCache == null)
             {
                 // cache has been GC'd,
                 destroy();
                 return;
             }
             // Members that have left the cluster and re-joined may be an inconsistent state due to the
             // potential for "split brain".  So err on the side of safety and flush the local cache.
             localCache.removeAll();
         }
 
         @Override
         public void onMessage(Message<K> message) {
             Cache<K, ?> localCache = localCacheRef.get();
             if (localCache == null)
             {
                 // cache has been gc-d
                 destroy();
                 return;
             }
             if (!message.getPublishingMember().localMember())
             {
                 K key = message.getMessageObject();
                 if (key == null)
                 {
                     localCache.removeAll();
                 }
                 else
                 {
                     localCache.remove(key);
                 }
             }
         }
 
         void destroy()
         {
             cluster.removeMembershipListener(membershipListenerId);
             topic.removeMessageListener(topicListenerId);
         }
 
         void publish(K message)
         {
             topic.publish(message);
         }
     }
 }