package com.atlassian.cache.hazelcast;
   
   import javax.annotation.Nonnull;
   
   import com.atlassian.cache.CacheException;
   import com.atlassian.cache.CacheFactory;
   import com.atlassian.cache.CacheSettings;
   import com.atlassian.cache.CachedReference;
   import com.atlassian.cache.CachedReferenceEvent;
  import com.atlassian.cache.CachedReferenceListener;
  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.EntryAdapter;
  import com.hazelcast.core.EntryEvent;
  import com.hazelcast.core.IMap;
  
  /**
   * 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.
   *
   * @since 2.4.0
   */
  public class HazelcastHybridCachedReference<V> extends ManagedHybridCacheSupport implements CachedReference<V>
  {
      private final CachedReference<Versioned<V>> localReference;
      private final IMap<ReferenceKey, Long> versionMap;
  
      private final CachedReferenceListenerSupport<V> listenerSupport = new ValueCachedReferenceListenerSupport<V>()
      {
          @Override
          protected void init(CachedReferenceListenerSupport<V> actualListenerSupport)
          {
              versionMap.addEntryListener(new HazelcastHybridReferenceEntryListener(), false);
          }
  
          @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);
          }
      };
  
      public HazelcastHybridCachedReference(String name, CacheFactory localFactory, final IMap<ReferenceKey, Long> versionMap,
              final Supplier<V> supplier, HazelcastCacheManager cacheManager)
      {
          super(name, cacheManager);
  
          Supplier<Versioned<V>> localSupplier = new Supplier<Versioned<V>>()
          {
              @Override
              public Versioned<V> get()
              {
                  long version = getVersion();
                  V value = supplier.get();
                  if (value == null)
                  {
                      throw new CacheException("The Supplier for cached reference '" + getName() + "'returned null. Null values are not supported.");
                  }
                  return new Versioned<V>(value, version);
              }
          };
          this.versionMap = versionMap;
          this.localReference = localFactory.getCachedReference(name, localSupplier, getCacheSettings());
      }
  
      @Nonnull
      @Override
      public V get()
      {
          Versioned<V> value = localReference.get();
          Long version = versionMap.get(ReferenceKey.KEY);
          if (version == null || value.getVersion() != version)
          {
              // version mismatch
              localReference.reset();
              value = localReference.get(); // will trigger a new call to supplier to re-generate the value
          }
          return value.getValue();
      }
  
      @Override
      public boolean isFlushable()
      {
          return getCacheSettings().getFlushable(true);
      }
  
      @Override
      public boolean isReplicateAsynchronously()
      {
         return false;
     }
 
     @Override
     public void reset()
     {
         versionMap.executeOnKey(ReferenceKey.KEY, IncrementVersionEntryProcessor.getInstance());
         localReference.reset();
     }
 
     @Override
     protected ManagedCache getLocalCache()
     {
         return (ManagedCache) localReference;
     }
 
     private CacheSettings getCacheSettings()
     {
         return cacheManager.getCacheSettings(getHazelcastMapName());
     }
 
     private String getHazelcastMapName()
     {
         return versionMap.getName();
     }
 
     @Override
     public void clear()
     {
         if (isFlushable())
         {
             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 long getVersion()
     {
         // try a standard get to give the near-cache a chance
         Long version = versionMap.get(ReferenceKey.KEY);
         if (version == null)
         {
             version = (Long) versionMap.executeOnKey(ReferenceKey.KEY, GetOrInitVersionEntryProcessor.getInstance());
         }
         return version;
     }
 
     private class HazelcastHybridReferenceEntryListener extends EntryAdapter<ReferenceKey, Long>
     {
         @Override
         public void entryRemoved(EntryEvent<ReferenceKey, Long> event)
         {
             // This should not happen because on all operations we just bump up the shared version.
             localReference.reset();
         }
 
         @Override
         public void entryUpdated(EntryEvent<ReferenceKey, Long> event)
         {
             // The only mechanism we are employing here is version bump. But we don't support set/put
             // so a version bump always means reset.
             localReference.reset();
         }
 
         @Override
         public void entryEvicted(EntryEvent<ReferenceKey, Long> event)
         {
             // Eviction of the shared value should bring eviction on the local value. However we don't
             // have access to this. Probably the best we can do is to trigger reset - probably will result in
             // two events - evict and reset.
             localReference.reset();
         }
     }
 
     private static class DelegatingCachedReferenceListener<V> implements CachedReferenceListener<Versioned<V>>
     {
         private final CachedReferenceListenerSupport<V> listenerSupport;
 
         private DelegatingCachedReferenceListener(final CachedReferenceListenerSupport<V> listenerSupport)
         {
             this.listenerSupport = listenerSupport;
         }
 
         @Override
         public void onEvict(@Nonnull CachedReferenceEvent<Versioned<V>> event)
         {
             listenerSupport.notifyEvict(get(event.getValue()));
         }
 
         @Override
         public void onSet(@Nonnull CachedReferenceEvent<Versioned<V>> event)
         {
             listenerSupport.notifySet(get(event.getValue()));
         }
 
         @Override
         public void onReset(@Nonnull CachedReferenceEvent<Versioned<V>> event)
         {
             listenerSupport.notifyReset(get(event.getValue()));
         }
 
         private V get(Versioned<V> versioned)
         {
             return versioned != null ? versioned.getValue() : null;
         }
     }
 }