package com.atlassian.cache.memory;
   
   import java.util.SortedMap;
   import java.util.concurrent.Callable;
   
   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.SimpleTimer;
  import com.atlassian.instrumentation.caches.CacheCollector;
  import com.atlassian.instrumentation.caches.CacheKeys;
  
  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 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>
  {
      private final LoadingCache<ReferenceKey, V> internalCache;
      private final CachedReferenceListenerSupport<V> listenerSupport;
      private final CacheCollector counter;
  
      private 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.counter = collector;
      }
  
      static <V> DelegatingCachedReference<V> create(final LoadingCache<ReferenceKey, V> internalCache,
              String name, CacheSettings settings, CacheCollector collector)
      {
          return new DelegatingCachedReference<V>(internalCache, name, settings, collector);
      }
  
      @Override
      public CacheCollector getCacheCollector()
      {
          return counter;
      }
  
      @Nonnull
      @Override
      public V get()
      {
          try
          {
              V value = internalCache.getIfPresent(ReferenceKey.KEY);
              if (value != null)
              {
                  if (isStatisticsEnabled())
                  {
                      counter.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 (isStatisticsEnabled())
             {
                 counter.remove();
             }
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
     }
 
     @Override
     public void clear()
     {
         reset();
     }
 
     @Override
     public void setStatistics(final boolean enabled)
     {
         counter.setEnabled(enabled);
     }
 
     @Override
     public boolean isStatisticsEnabled()
     {
         return counter.isEnabled();
     }
 
     @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, com.atlassian.util.concurrent.Supplier<Long>> getStatistics()
     {
         if (isStatisticsEnabled())
         {
             return toStatistics(counter);
         }
         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;
         }
     }
 }