package com.atlassian.cache.vcache;
   
   import java.io.Serializable;
   import java.util.Collection;
   import java.util.Optional;
   import java.util.concurrent.CompletionException;
   import javax.annotation.Nonnull;
   import javax.annotation.Nullable;
   
  import com.atlassian.cache.Cache;
  import com.atlassian.cache.CacheEntryListener;
  import com.atlassian.cache.CacheException;
  import com.atlassian.cache.CacheLoader;
  import com.atlassian.cache.CacheSettings;
  import com.atlassian.cache.Supplier;
  import com.atlassian.vcache.DirectExternalCache;
  import com.atlassian.vcache.IdentifiedValue;
  import com.atlassian.vcache.PutPolicy;
  
  import static com.atlassian.cache.vcache.Utils.asSerializable;
  import static com.atlassian.cache.vcache.Utils.asString;
  import static com.atlassian.vcache.VCacheUtils.fold;
  import static com.atlassian.vcache.VCacheUtils.join;
  import static java.util.Objects.requireNonNull;
  
  class ExternalCacheDelegate<K, V>
          extends ManagedCacheSupport
          implements Cache<K, V>
  {
      private final DirectExternalCache<Serializable> delegate;
      private final CacheLoader<String, Serializable> cacheLoader;
  
      public ExternalCacheDelegate(
              String name,
              DirectExternalCache<Serializable> delegate,
              @Nullable CacheLoader<String, Serializable> cacheLoader,
              CacheSettings settings)
      {
          super(name, settings);
          this.delegate = requireNonNull(delegate);
          this.cacheLoader = cacheLoader;
      }
  
      @Override
      public void clear()
      {
          join(delegate.removeAll());
      }
  
      @Override
      public boolean isLocal()
      {
          return false;
      }
  
      @Override
      public boolean containsKey(@Nonnull K key)
      {
          return join(delegate.get(asString(key))).isPresent();
      }
  
      @Nonnull
      @Override
      public Collection<K> getKeys()
      {
          throw new UnsupportedOperationException("Not supported on non-local caches");
      }
  
      @Nullable
      @Override
      public V get(@Nonnull K key)
      {
          final String strKey = asString(key);
          try
          {
              if (cacheLoader == null)
              {
                  return (V) join(delegate.get(strKey)).orElse(null);
              }
              else
              {
                  return (V) join(delegate.get(strKey, () -> asSerializable(cacheLoader.load(strKey))));
              }
          }
          catch (CompletionException ce)
          {
              throw new CacheException(ce);
          }
      }
  
      @Nonnull
      @Override
      public V get(@Nonnull K key, @Nonnull Supplier<? extends V> valueSupplier)
      {
          return (V) join(delegate.get(asString(key), () -> asSerializable(valueSupplier.get())));
      }
  
      @Override
      public void put(@Nonnull K key, @Nonnull V value)
     {
         fold(delegate.put(asString(key), asSerializable(value), PutPolicy.PUT_ALWAYS),
                 success -> success,
                 failure -> {
                     throw new CacheException("put() operation failed", failure);
                 });
     }
 
     @Nullable
     @Override
     public V putIfAbsent(@Nonnull K key, @Nonnull V value)
     {
         for (; ; )
         {
             final Optional<Serializable> current = join(delegate.get(asString(key)));
             if (current.isPresent())
             {
                 return (V) current.get();
             }
             else
             {
                 final boolean added = join(delegate.put(asString(key), asSerializable(value), PutPolicy.ADD_ONLY));
                 if (added)
                 {
                     return null; // We are happy chaps
                 }
             }
         }
     }
 
     @Override
     public void remove(@Nonnull K key)
     {
         join(delegate.remove(asString(key)));
     }
 
     @Override
     public boolean remove(@Nonnull K key, @Nonnull V value)
     {
         for (; ; )
         {
             final Optional<IdentifiedValue<Serializable>> current = join(delegate.getIdentified(asString(key)));
             if (current.isPresent() && current.get().value().equals(value))
             {
                 final boolean removed = join(delegate.removeIf(asString(key), current.get().identifier()));
                 if (removed)
                 {
                     return true;
                 }
                 // Time to loop and try again
             }
             else
             {
                 return false; // Not a matching value
             }
         }
     }
 
     @Override
     public void removeAll()
     {
         join(delegate.removeAll());
     }
 
     @Override
     public boolean replace(@Nonnull K key, @Nonnull V oldValue, @Nonnull V newValue)
     {
         for (; ; )
         {
             final Optional<IdentifiedValue<Serializable>> current = join(delegate.getIdentified(asString(key)));
             if (current.isPresent() && current.get().value().equals(oldValue))
             {
                 final boolean replaced = join(delegate.replaceIf(
                         asString(key), current.get().identifier(), asSerializable(newValue)));
                 if (replaced)
                 {
                     return true;
                 }
                 // Time to loop and try again
             }
             else
             {
                 return false; // Not a matching value
             }
         }
     }
 
     @Override
     public void addListener(@Nonnull CacheEntryListener<K, V> listener, boolean includeValues)
     {
         throw new UnsupportedOperationException("Not supported on non-local caches");
     }
 
     @Override
     public void removeListener(@Nonnull CacheEntryListener<K, V> listener)
     {
         throw new UnsupportedOperationException("Not supported on non-local caches");
     }
 }