package com.atlassian.cache.memory;
   
   import java.util.SortedMap;
   
   import javax.annotation.Nonnull;
   import javax.annotation.Nullable;
   
   import com.atlassian.cache.CacheException;
   import com.atlassian.cache.CacheSettings;
  import com.atlassian.cache.CacheStatisticsKey;
  import com.atlassian.cache.CachedReference;
  import com.atlassian.cache.CachedReferenceListener;
  import com.atlassian.cache.impl.CachedReferenceListenerSupport;
  import com.atlassian.cache.impl.DefaultCachedReferenceListenerSupport;
  import com.atlassian.cache.impl.ReferenceKey;
  import com.atlassian.instrumentation.caches.CacheCollector;
  
  import com.google.common.cache.LoadingCache;
  import com.google.common.cache.RemovalListener;
  import com.google.common.cache.RemovalNotification;
  import com.google.common.collect.ImmutableSortedMap;
  import com.google.common.util.concurrent.UncheckedExecutionException;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import static com.atlassian.cache.memory.DelegatingCacheStatistics.toStatistics;
  
  /**
   * A Lazy Reference that delegates Concurrent Map.
   *
   * @since 2.0
   */
  class DelegatingCachedReference<V> extends ManagedCacheSupport implements CachedReference<V>
  {
      static final CreateFunction DEFAULT_CREATE_FUNCTION = new DefaultCreateFunction();
  
      private final Logger eventLogger;
      private final Logger stacktraceLogger;
  
      private final LoadingCache<ReferenceKey, V> internalCache;
      private final CachedReferenceListenerSupport<V> listenerSupport;
      private final CacheCollector collector;
  
      protected DelegatingCachedReference(final LoadingCache<ReferenceKey, V> internalCache,
              String name, CacheSettings settings, final CacheCollector collector)
      {
          super(name, settings);
          this.internalCache = internalCache;
  
          this.listenerSupport = new DefaultCachedReferenceListenerSupport<V>();
          this.collector = collector;
  
          this.eventLogger = LoggerFactory.getLogger("com.atlassian.cache.event." + name);
          this.stacktraceLogger = LoggerFactory.getLogger("com.atlassian.cache.stacktrace." + name);
      }
  
      static <V> DelegatingCachedReference<V> create(final LoadingCache<ReferenceKey, V> internalCache,
              String name, CacheSettings settings, CacheCollector collector)
      {
          return DEFAULT_CREATE_FUNCTION.create(internalCache, name, settings, collector);
      }
  
      @Override
      public CacheCollector getCacheCollector()
      {
          return collector;
      }
  
      @Nonnull
      @Override
      public V get()
      {
          try
          {
              V value = internalCache.getIfPresent(ReferenceKey.KEY);
              if (value != null)
              {
                  if (isCollectorStatisticsEnabled())
                  {
                      collector.hit();
                  }
                  return value;
              }
              else
              {
                  return getUnderLock();
              }
          }
          catch (UncheckedExecutionException e)
          {
              throw new CacheException(e.getCause());
          }
          catch (Exception e)
          {
              throw new CacheException(e);
          }
      }
  
      synchronized private V getUnderLock()
     {
         return internalCache.getUnchecked(ReferenceKey.KEY);
     }
 
     @Override
     synchronized public void reset()
     {
         try
         {
             internalCache.invalidate(ReferenceKey.KEY);
             if (isCollectorStatisticsEnabled())
             {
                 collector.remove();
             }
             eventLogger.info("Cache {} was flushed", getName());
             if (stacktraceLogger.isInfoEnabled()) {
                 stacktraceLogger.info("Cache {} was flushed. Stacktrace:", getName(), new Exception());
             }
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
     }
 
     @Override
     public void clear()
     {
         reset();
     }
 
     private boolean isCollectorStatisticsEnabled()
     {
         return collector.isEnabled();
     }
 
     @Override
     public boolean isStatisticsEnabled() {
         return settings.getStatisticsEnabled();
     }
 
     @Override
     public boolean equals(@Nullable final Object other)
     {
         if (other instanceof DelegatingCachedReference)
         {
             DelegatingCachedReference<?> otherDelegatingReference = (DelegatingCachedReference<?>) other;
             if (internalCache.equals(otherDelegatingReference.internalCache))
             {
                 return true;
             }
         }
         return false;
     }
 
     @Override
     public int hashCode()
     {
         return 3 + internalCache.hashCode();
     }
 
     @Nonnull
     @Override
     public SortedMap<CacheStatisticsKey, java.util.function.Supplier<Long>> getStatistics()
     {
         if (isStatisticsEnabled())
         {
             return toStatistics(internalCache);
         }
         else
         {
             return ImmutableSortedMap.of();
         }
     }
 
     @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);
     }
 
     protected static class DelegatingReferenceRemovalListener<V> implements RemovalListener<ReferenceKey, V>
     {
         private DelegatingCachedReference<V> cachedReference;
 
         protected void onSupply(V value)
         {
             cachedReference.listenerSupport.notifySet(value);
         }
 
         @Override
         public void onRemoval(RemovalNotification<ReferenceKey, V> notification)
         {
             switch (notification.getCause())
             {
                 case COLLECTED:
                 case EXPIRED:
                     cachedReference.listenerSupport.notifyEvict(notification.getValue());
                     break;
                 case EXPLICIT:
                     cachedReference.listenerSupport.notifyReset(notification.getValue());
                     break;
                 case REPLACED:
                     V value = cachedReference.internalCache.getIfPresent(ReferenceKey.KEY);
                     cachedReference.listenerSupport.notifySet(value);
                     break;
             }
         }
 
         public void setCachedReference(DelegatingCachedReference<V> cachedReference)
         {
             this.cachedReference = cachedReference;
         }
     }
 
     public interface CreateFunction
     {
         <V> DelegatingCachedReference<V> create(final LoadingCache<ReferenceKey, V> internalCache,
                                                 String name, CacheSettings settings, CacheCollector collector);
     }
 
     public static class DefaultCreateFunction implements CreateFunction
     {
 
         @Override
         public <V> DelegatingCachedReference<V> create(final LoadingCache<ReferenceKey, V> internalCache,
                                                   String name, CacheSettings settings, CacheCollector collector) {
             return new DelegatingCachedReference<>(internalCache, name, settings, collector);
         }
     }
 }