package com.atlassian.vcache.internal.core.service;
   
   import com.atlassian.vcache.PutPolicy;
   import com.atlassian.vcache.StableReadExternalCache;
   import com.atlassian.vcache.VCacheException;
   import com.atlassian.vcache.internal.core.metrics.MetricsRecorder;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   
  import java.util.Map;
  import java.util.Optional;
  import java.util.Set;
  import java.util.concurrent.CompletionStage;
  import java.util.concurrent.locks.ReentrantLock;
  import java.util.function.Function;
  import java.util.function.Supplier;
  import java.util.stream.Collectors;
  import java.util.stream.StreamSupport;
  
  /**
   * Provides operations common to {@link com.atlassian.vcache.StableReadExternalCache} instances.
   * <p>
   * Locking is provided to synchronise the local and remote caches. There are two locking strategies - a cache wide
   * read-write lock and a per-key lock. The cache wide lock provides exclusion for multi-key operations such as
   * removeAll and getBulk. The key lock provides locking for single key get and put whilst honouring the cache lock.
   *
   * @param <V> the value type
   * @since 1.0.0
   */
  public abstract class AbstractStableReadExternalCache<V>
          extends AbstractNonDirectExternalCache<V>
          implements StableReadExternalCache<V> {
  
      private static final Logger log = LoggerFactory.getLogger(AbstractStableReadExternalCache.class);
  
      private final ReentrantLock cacheLock = new ReentrantLock();
  
      protected AbstractStableReadExternalCache(String name, MetricsRecorder metricsRecorder) {
          super(name, metricsRecorder);
      }
  
      protected final Map<String, Optional<V>> checkValuesRecorded(Iterable<String> internalKeys) {
          final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
          //noinspection OptionalGetWithoutIsPresent
          return StreamSupport.stream(internalKeys.spliterator(), false)
                  .filter(k -> cacheContext.getValueRecorded(k).isPresent())
                  .distinct()
                  .collect(Collectors.toMap(
                          k -> k,
                          k -> cacheContext.getValueRecorded(k).get())
                  );
      }
  
      @Override
      public final CompletionStage<Optional<V>> get(String internalKey) {
          return withCacheLock(() -> super.get(internalKey));
      }
  
      @Override
      public final CompletionStage<V> get(String internalKey, Supplier<V> supplier) {
          return withCacheLock(() -> super.get(internalKey, supplier));
      }
  
      @Override
      public final CompletionStage<Map<String, Optional<V>>> getBulk(Iterable<String> internalKeys) {
          return withCacheLock(() -> super.getBulk(internalKeys));
      }
  
      @Override
      public final CompletionStage<Map<String, V>> getBulk(Function<Set<String>, Map<String, V>> factory, Iterable<String> internalKeys) {
          return withCacheLock(() -> super.getBulk(factory, internalKeys));
      }
  
      @Override
      public final CompletionStage<Boolean> put(final String internalKey, final V value, final PutPolicy policy) {
          return withCacheLock(() -> internalPut(internalKey, value, policy));
      }
  
      protected abstract CompletionStage<Boolean> internalPut(String internalKey, V value, PutPolicy policy);
  
      @Override
      public final CompletionStage<Void> remove(final Iterable<String> keys) {
          return withCacheLock(() -> internalRemove(keys));
      }
  
      protected abstract CompletionStage<Void> internalRemove(Iterable<String> keys);
  
  
      @Override
      public CompletionStage<Void> removeAll() {
          return withCacheLock(this::internalRemoveAll);
      }
  
      protected abstract CompletionStage<Void> internalRemoveAll();
  
      private <R> R withCacheLock(final Supplier<R> result) {
          if (cacheLock.isHeldByCurrentThread()) {
              log.error("Detected a recursive call to cache: {}", getName());
              throw new VCacheException("Detected a recursive call to cache: " + getName());
         }
         cacheLock.lock();
         try {
             return result.get();
         } finally {
             cacheLock.unlock();
         }
     }
 }