package com.atlassian.cache.hazelcast;
   
   import javax.annotation.Nonnull;
   
   import com.atlassian.cache.CachedReference;
   import com.atlassian.cache.CachedReferenceEvent;
   import com.atlassian.cache.CachedReferenceListener;
   import com.atlassian.cache.CacheFactory;
   import com.atlassian.cache.CacheSettings;
  import com.atlassian.cache.ManagedCache;
  import com.atlassian.cache.Supplier;
  import com.atlassian.cache.impl.CachedReferenceListenerSupport;
  import com.atlassian.cache.impl.ReferenceKey;
  import com.atlassian.cache.impl.ValueCachedReferenceListenerSupport;
  
  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;
  
  import java.lang.ref.WeakReference;
  
  /**
   * Implementation of {@link ManagedCache} and {@link com.atlassian.cache.CachedReference} that can be used when the
   * cached value does not implement {@code Serializable} but reference invalidation must work cluster-wide. It works by
   * caching the value locally and broadcasting invalidations cluster-wide using a Hazelcast {@link ITopic}.
   *
   * @since 2.12.0
   */
  public class HazelcastAsyncHybridCachedReference<V> extends ManagedHybridCacheSupport implements CachedReference<V>
  {
      private final AsyncInvalidationListener listener;
      private final CachedReferenceListenerSupport<V> listenerSupport = new ValueCachedReferenceListenerSupport<V>()
      {
          @Override
          protected void initValue(final CachedReferenceListenerSupport<V> actualListenerSupport)
          {
              localReference.addListener(new DelegatingCachedReferenceListener<V>(actualListenerSupport), true);
          }
  
          @Override
          protected void initValueless(final CachedReferenceListenerSupport<V> actualListenerSupport)
          {
              localReference.addListener(new DelegatingCachedReferenceListener<V>(actualListenerSupport), false);
          }
      };
      private final CachedReference<V> localReference;
  
      public HazelcastAsyncHybridCachedReference(String name, CacheFactory localFactory, final ITopic<ReferenceKey> topic,
              final Supplier<V> supplier, HazelcastCacheManager cacheManager, CacheSettings settings)
      {
          super(name, cacheManager);
          this.localReference = localFactory.getCachedReference(name, supplier, settings);
          listener = new AsyncInvalidationListener(cacheManager.getHazelcastInstance().getCluster(), localReference,
                  topic);
      }
  
      @Nonnull
      @Override
      public V get()
      {
          return localReference.get();
      }
  
      @Override
      public boolean isFlushable()
      {
          return true;
      }
  
      @Override
      public boolean isReplicateAsynchronously()
      {
          return true;
      }
  
      @Override
      public void reset()
      {
          localReference.reset();
          invalidateRemotely();
      }
  
      @Override
      protected ManagedCache getLocalCache()
      {
          return (ManagedCache) localReference;
      }
  
      @Override
      public void clear()
      {
          reset();
      }
  
      @Override
      public boolean updateMaxEntries(int newValue)
     {
         return false;
     }
 
     @Override
     public void addListener(@Nonnull CachedReferenceListener<V> listener, boolean includeValues)
     {
         listenerSupport.add(listener, includeValues);
     }
 
     @Override
     public void removeListener(@Nonnull CachedReferenceListener<V> listener)
     {
         listenerSupport.remove(listener);
     }
 
     private void invalidateRemotely()
     {
         listener.publish(null);
     }
 
     private static class DelegatingCachedReferenceListener<V> implements CachedReferenceListener<V>
     {
         private final CachedReferenceListenerSupport<V> listenerSupport;
 
         private DelegatingCachedReferenceListener(final CachedReferenceListenerSupport<V> listenerSupport)
         {
             this.listenerSupport = listenerSupport;
         }
 
         @Override
         public void onEvict(@Nonnull CachedReferenceEvent<V> event)
         {
             listenerSupport.notifyEvict(event.getValue());
         }
 
         @Override
         public void onSet(@Nonnull CachedReferenceEvent<V> event)
         {
             listenerSupport.notifySet(event.getValue());
         }
 
         @Override
         public void onReset(@Nonnull CachedReferenceEvent<V> event)
         {
             listenerSupport.notifyReset(event.getValue());
         }
     }
 
     private static class AsyncInvalidationListener extends MembershipAdapter implements MessageListener<ReferenceKey> {
 
         private final Cluster cluster;
         private final WeakReference<CachedReference<?>> localReferenceRef;
         private final String membershipListenerId;
         private final ITopic<ReferenceKey> topic;
         private final String topicListenerId;
 
         AsyncInvalidationListener(Cluster cluster, CachedReference<?> localReference, ITopic<ReferenceKey> topic)
         {
             this.cluster = cluster;
             this.localReferenceRef = new WeakReference<>(localReference);
             this.topic = topic;
             this.topicListenerId = topic.addMessageListener(this);
             this.membershipListenerId = cluster.addMembershipListener(this);
         }
 
         @Override
         public void memberAdded(MembershipEvent membershipEvent)
         {
             CachedReference<?> localReference = localReferenceRef.get();
             if (localReference == 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.
             localReference.reset();
         }
 
         @Override
         public void onMessage(Message<ReferenceKey> message) {
             CachedReference<?> localReference = localReferenceRef.get();
             if (localReference == null)
             {
                 // cache has been gc-d
                 destroy();
                 return;
             }
             if (!message.getPublishingMember().localMember())
             {
                 localReference.reset();
             }
         }
 
         void destroy()
         {
             cluster.removeMembershipListener(membershipListenerId);
             topic.removeMessageListener(topicListenerId);
         }
 
         void publish(ReferenceKey message)
         {
             topic.publish(message);
         }
     }
 }