package com.atlassian.cache.ehcache;
   
   import com.atlassian.cache.Cache;
   import com.atlassian.cache.CacheEntryListener;
   import com.atlassian.cache.CacheException;
   import com.atlassian.cache.CacheSettings;
   import com.atlassian.cache.CacheStatisticsKey;
   import com.atlassian.cache.ehcache.wrapper.ValueProcessor;
   import com.atlassian.cache.ehcache.wrapper.ValueProcessorEhcacheLoaderDecorator;
  import com.atlassian.cache.impl.CacheEntryListenerSupport;
  import com.atlassian.cache.impl.LazyCacheEntryListenerSupport;
  import com.google.common.collect.ImmutableSortedMap;
  import net.sf.ehcache.Ehcache;
  import net.sf.ehcache.Element;
  import net.sf.ehcache.event.CacheEventListener;
  import net.sf.ehcache.loader.CacheLoader;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import javax.annotation.Nonnull;
  import javax.annotation.Nullable;
  import java.util.Collection;
  import java.util.SortedMap;
  import java.util.function.Supplier;
  
  import static com.atlassian.cache.ehcache.DelegatingCacheStatistics.toStatistics;
  import static com.atlassian.cache.ehcache.wrapper.WrapperUtils.unwrapAllKeys;
  import static com.atlassian.cache.ehcache.wrapper.WrapperUtils.unwrapElement;
  
  /**
   * A Cache that delegates to EhCache.
   *
   * @since 2.0
   */
  class DelegatingCache<K, V> extends ManagedCacheSupport implements Cache<K, V>
  {
      private final Ehcache delegate;
      private final Logger eventLogger;
      private final Logger stacktraceLogger;
  
      private final CacheEntryListenerSupport<K, V> listenerSupport = new LazyCacheEntryListenerSupport<K, V>()
      {
          @Override
          protected void init()
          {
              delegate.getCacheEventNotificationService().registerListener(new DelegatingCacheEventListener());
          }
      };
  
  
      private final ValueProcessor valueProcessor;
  
      private DelegatingCache(final Ehcache delegate, CacheSettings settings, final ValueProcessor valueProcessor)
      {
          super(delegate, settings);
          this.delegate = delegate;
          this.eventLogger = LoggerFactory.getLogger("com.atlassian.cache.event." + delegate.getName());
          this.stacktraceLogger = LoggerFactory.getLogger("com.atlassian.cache.stacktrace." + delegate.getName());
          this.valueProcessor = valueProcessor;
      }
  
      static <K, V> DelegatingCache<K, V> create(final Ehcache delegate, CacheSettings settings, final ValueProcessor valueProcessor)
      {
          return new DelegatingCache<>(delegate, settings, valueProcessor);
      }
  
      @Override
      public boolean containsKey(@Nonnull K key)
      {
          return delegate.isKeyInCache(wrap(key));
      }
  
      @Nonnull
      @SuppressWarnings("unchecked")
      @Override
      public Collection<K> getKeys()
      {
          try
          {
              return unwrapAllKeys(delegate.getKeys(), valueProcessor);
          }
          catch (Exception e)
          {
              throw new CacheException(e);
          }
      }
  
      @Override
      public void put(@Nonnull final K key, @Nonnull final V value)
      {
          try
          {
              delegate.put(new Element(wrap(key), wrap(value)));
          }
          catch (Exception e)
          {
              throw new CacheException(e);
          }
      }
 
     @SuppressWarnings("unchecked")
     @Nullable
     @Override
     public V get(@Nonnull final K key)
     {
         try
         {
             Element element = unwrap(delegate.get(wrap(key)));
             return element == null ? null : (V) element.getObjectValue();
         }
         catch (net.sf.ehcache.CacheException e)
         {
             throw new CacheException(e.getCause());
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
     }
 
     @SuppressWarnings("unchecked")
     @Nonnull
     @Override
     public V get(@Nonnull final K key, @Nonnull final com.atlassian.cache.Supplier<? extends V> valueSupplier)
     {
         try
         {
             Element element = unwrap(delegate.getWithLoader(wrap(key), getCacheLoader(valueSupplier), null));
             return (V) element.getObjectValue();
         }
         catch (net.sf.ehcache.CacheException e)
         {
             throw new CacheException(e.getCause());
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
     }
 
     @Override
     public void remove(@Nonnull final K key)
     {
         try
         {
             delegate.remove(wrap(key));
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
     }
 
     @Override
     public void removeAll()
     {
         try
         {
             delegate.removeAll();
             eventLogger.info("Cache {} was flushed", delegate.getName());
             if (stacktraceLogger.isInfoEnabled()) {
                 stacktraceLogger.info("Cache {} was flushed. Stacktrace:", delegate.getName(), new Exception());
             }
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
     }
 
     @Override
     public void clear()
     {
         removeAll();
     }
 
     @Nullable
     @Override
     @SuppressWarnings("unchecked")
     public V putIfAbsent(@Nonnull K key, @Nonnull V value)
     {
         try
         {
             Element previous = unwrap(delegate.putIfAbsent(new Element(wrap(key), wrap(value))));
             if (previous != null)
             {
                 return (V) previous.getObjectValue();
             }
             else
             {
                 return null;
             }
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
     }
 
     @Override
     public boolean remove(@Nonnull K key, @Nonnull V value)
     {
         try
         {
             return delegate.removeElement(new Element(wrap(key), wrap(value)));
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
     }
 
     @Override
     public boolean replace(@Nonnull K key, @Nonnull V oldValue, @Nonnull V newValue)
     {
         try
         {
             return delegate.replace(new Element(wrap(key), wrap(oldValue)), new Element(wrap(key), wrap(newValue)));
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
     }
 
     @Nonnull
     @Override
     public SortedMap<CacheStatisticsKey,Supplier<Long>> getStatistics()
     {
         if (isStatisticsEnabled())
         {
             return toStatistics(delegate.getStatistics());
         }
         else
         {
             return ImmutableSortedMap.of();
         }
     }
 
     @Override
     public boolean equals(@Nullable final Object other)
     {
         if (other instanceof DelegatingCache)
         {
             DelegatingCache<?,?> otherDelegatingCache = (DelegatingCache<?,?>)other;
             if (delegate.equals(otherDelegatingCache.delegate))
             {
                 return true;
             }
         }
         return false;
     }
 
     @Override
     public int hashCode()
     {
         return 3 + delegate.hashCode();
     }
 
     @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);
     }
 
     private Object wrap(Object o) {
         return valueProcessor.wrap(o);
     }
 
     private Object unwrap(Object o) {
         return valueProcessor.unwrap(o);
     }
 
     private Element unwrap(Element element) {
         return unwrapElement(element, valueProcessor);
     }
 
     private CacheLoader getCacheLoader(final com.atlassian.cache.Supplier<? extends V> valueSupplier) {
             return new ValueProcessorEhcacheLoaderDecorator(new ReferenceCacheLoader(valueSupplier), valueProcessor);
     }
 
     private class DelegatingCacheEventListener implements CacheEventListener
     {
         @Override
         public void notifyElementRemoved(Ehcache ehcache, Element element) throws net.sf.ehcache.CacheException
         {
             listenerSupport.notifyRemove((K) unwrap(element.getObjectKey()), (V) unwrap(element.getObjectValue()));
         }
 
         @Override
         public void notifyElementPut(Ehcache ehcache, Element element) throws net.sf.ehcache.CacheException
         {
             listenerSupport.notifyAdd((K) unwrap(element.getObjectKey()), (V) unwrap(element.getObjectValue()));
         }
 
         @Override
         public void notifyElementUpdated(Ehcache ehcache, Element element) throws net.sf.ehcache.CacheException
         {
             listenerSupport.notifyUpdate((K) unwrap(element.getObjectKey()), (V) unwrap(element.getObjectValue()), null);
         }
 
         @Override
         public void notifyElementExpired(Ehcache ehcache, Element element)
         {
             listenerSupport.notifyEvict((K) unwrap(element.getObjectKey()), (V) unwrap(element.getObjectValue()));
         }
 
         @Override
         public void notifyElementEvicted(Ehcache ehcache, Element element)
         {
             listenerSupport.notifyEvict((K) unwrap(element.getObjectKey()), (V) unwrap(element.getObjectValue()));
         }
 
         @Override
         public void notifyRemoveAll(Ehcache ehcache)
         {
             // There is no way to enumerate the keys that were in the cache, therefore we cannot
             // produce any meaningful event
         }
 
         @Override
         public void dispose()
         {
             // We don't hold onto any resources so there is nothing to be done.
         }
 
         public Object clone() throws CloneNotSupportedException
         {
             throw new CloneNotSupportedException();
         }
     }
 }