package com.atlassian.vcache.internal.memcached;
   
   import java.util.HashMap;
   import java.util.Map;
   import java.util.Optional;
   import java.util.Set;
   import java.util.concurrent.CompletableFuture;
   import java.util.concurrent.ExecutionException;
   import java.util.function.Function;
  import java.util.function.Supplier;
  import java.util.stream.Collectors;
  import java.util.stream.StreamSupport;
  import javax.annotation.Nonnull;
  
  import com.atlassian.vcache.ExternalCache;
  import com.atlassian.vcache.ExternalCacheSettings;
  import com.atlassian.vcache.Marshaller;
  import com.atlassian.vcache.MarshallerException;
  import com.atlassian.vcache.internal.RequestContext;
  import com.atlassian.vcache.internal.core.ExternalCacheKeyGenerator;
  import com.atlassian.vcache.internal.core.VCacheUtils;
  
  import net.spy.memcached.MemcachedClientIF;
  import net.spy.memcached.OperationTimeoutException;
  import org.slf4j.Logger;
  
  import static com.atlassian.vcache.internal.NameValidator.requireValidCacheName;
  import static com.atlassian.vcache.internal.memcached.MemcachedUtils.isEmpty;
  import static com.atlassian.vcache.internal.memcached.MemcachedUtils.perform;
  import static com.atlassian.vcache.internal.memcached.MemcachedUtils.unmarshall;
  import static java.util.Objects.requireNonNull;
  
  abstract class AbstractStableReadExternalCache<V> implements ExternalCache<V>
  {
      protected final Supplier<MemcachedClientIF> clientSupplier;
      protected final Supplier<RequestContext> contextSupplier;
      protected final ExternalCacheKeyGenerator keyGenerator;
      protected final String name;
      protected final Marshaller<V> valueMarshaller;
      protected final ExternalCacheSettings settings;
      protected final int defaultTtl;
  
      AbstractStableReadExternalCache(
              Supplier<MemcachedClientIF> clientSupplier,
              Supplier<RequestContext> contextSupplier,
              ExternalCacheKeyGenerator keyGenerator,
              String name,
              Marshaller<V> valueMarshaller,
              ExternalCacheSettings settings)
      {
          this.clientSupplier = requireNonNull(clientSupplier);
          this.contextSupplier = requireNonNull(contextSupplier);
          this.keyGenerator = requireNonNull(keyGenerator);
          this.name = requireValidCacheName(name);
          this.valueMarshaller = requireNonNull(valueMarshaller);
          this.settings = requireNonNull(settings);
          this.defaultTtl = VCacheUtils.roundUpToSeconds(settings.getDefaultTtl().get());
      }
  
      /**
       * Returns the cache context for the current request.
       * @return the cache context for the current request.
       * @throws OperationTimeoutException
       */
      @Nonnull
      abstract StableReadRequestContext<V> ensureCacheContext() throws OperationTimeoutException;
  
      /**
       * Returns the logging instance for the implementation class.
       * @return the logging instance for the implementation class.
       */
      @Nonnull
      abstract Logger getLogger();
  
      /**
       * Handles the creation of an entry, if required.
       *
       * @param internalKey the internal key for the entry.
       * @param supplier called to create the value, if required
       * @return the value associated with the key.
       */
      @Nonnull
      abstract V handleCreation(String internalKey, Supplier<V> supplier)
              throws MarshallerException, ExecutionException, InterruptedException;
  
      /**
       * Handles the creation of a number of entries
       * @param factory called to create the required values.
       * @param externalKeys the external keys for the entries.
       *
       * @return the values associated with the supplied keys (mapped on internal key)
       */
      @Nonnull
      abstract Map<String, V> handleCreation(
          Function<Set<String>, Map<String, V>> factory, Set<String> externalKeys)
              throws ExecutionException, InterruptedException;
  
      /**
       * Checks if a value is recorded for a specified internal key.
      */
     @Nonnull
     abstract Optional<Optional<V>> checkValueRecorded(String internalKey);
 
     /**
      * Checks for values recorded for specified internal keys.
      */
     @Nonnull
     abstract Map<String, Optional<V>> checkValuesRecorded(Iterable<String> internalKeys);
 
     @Nonnull
     @Override
     public final String getName()
     {
         return name;
     }
 
     @Nonnull
     @Override
     public final CompletableFuture<Optional<V>> get(String internalKey)
     {
         return perform(() -> {
             final StableReadRequestContext<V> cacheContext = ensureCacheContext();
 
             final Optional<Optional<V>> prior = checkValueRecorded(internalKey);
 
             return prior.orElseGet(() -> {
                 final String externalKey = cacheContext.externalEntryKeyFor(internalKey);
 
                 final Optional<V> result = unmarshall(clientSupplier.get().get(externalKey), valueMarshaller);
                 cacheContext.recordValue(internalKey, result);
                 return result;
             });
         });
     }
 
     @Nonnull
     @Override
     public final CompletableFuture<V> get(String internalKey, Supplier<V> supplier)
     {
         return perform(() -> {
             final StableReadRequestContext<V> cacheContext = ensureCacheContext();
 
             final Optional<Optional<V>> prior = checkValueRecorded(internalKey);
             if (prior.isPresent() && prior.get().isPresent())
             {
                 return prior.get().get();
             }
 
             if (prior.isPresent())
             {
                 if (prior.get().isPresent())
                 {
                     return prior.get().get();
                 }
                 else
                 {
                     getLogger().trace("Cache {}, creating candidate for key {}", name, internalKey);
                     return handleCreation(internalKey, supplier);
                 }
             }
 
             // Either way, there was no prior value recorded, or it was blank. So need to create the value
             // and write it back.
             final String externalKey = cacheContext.externalEntryKeyFor(internalKey);
             final Optional<V> result = unmarshall(clientSupplier.get().get(externalKey), valueMarshaller);
             if (result.isPresent())
             {
                 // A valid value exists in the external cache
                 cacheContext.recordValue(internalKey, result);
                 return result.get();
             }
 
             getLogger().trace("Cache {}, creating candidate for key {}", name, internalKey);
             return handleCreation(internalKey, supplier);
         });
     }
 
     @Nonnull
     @Override
     public final CompletableFuture<Map<String, Optional<V>>> getBulk(Iterable<String> internalKeys)
     {
         return perform(() -> {
             if (isEmpty(internalKeys))
             {
                 return new HashMap<>();
             }
 
             // Get the recorded values first
             final StableReadRequestContext<V> cacheContext = ensureCacheContext();
             final Map<String, Optional<V>> grandResult = checkValuesRecorded(internalKeys);
 
             // Calculate the externalKeys for the entries that are missing
             final Set<String> missingExternalKeys = StreamSupport.stream(internalKeys.spliterator(), false)
                     .filter(k -> !grandResult.containsKey(k))
                     .map(cacheContext::externalEntryKeyFor)
                     .collect(Collectors.toSet());
 
             if (missingExternalKeys.isEmpty())
             {
                 getLogger().trace("Cache {}: getBulk(): have all the requested entries cached", name);
                 return grandResult;
             }
             getLogger().trace("Cache {}: getBulk(): not cached {} requested entries", name, missingExternalKeys.size());
 
             // Get the missing values.
             // Returns map of keys that contain values, so need to handle the missing ones
             final Map<String, Object> haveValues = clientSupplier.get().getBulk(missingExternalKeys);
 
             return missingExternalKeys.stream().collect(
                     () -> grandResult,
                     (m, k) -> {
                         final Optional<V> result = unmarshall(haveValues.get(k), valueMarshaller);
                         cacheContext.recordValue(cacheContext.internalEntryKeyFor(k), result);
                         m.put(cacheContext.internalEntryKeyFor(k), result);
                     },
                     Map::putAll
             );
         });
     }
 
     @Nonnull
     @Override
     public final CompletableFuture<Map<String, V>> getBulk(
             Function<Set<String>, Map<String, V>> factory, Iterable<String> internalKeys)
     {
         return perform(() -> {
             if (isEmpty(internalKeys))
             {
                 return new HashMap<>();
             }
 
             // Get the recorded values first and collect the ones that have values.
             final StableReadRequestContext<V> cacheContext = ensureCacheContext();
             final Map<String, V> grandResult = checkValuesRecorded(internalKeys).entrySet().stream()
                     .filter(e -> e.getValue().isPresent())
                     .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().get()));
 
             // Calculate the candidate externalKeys for the entries that are missing
             final Set<String> candidateMissingExternalKeys = StreamSupport.stream(internalKeys.spliterator(), false)
                     .filter(k -> !grandResult.containsKey(k))
                     .map(cacheContext::externalEntryKeyFor)
                     .collect(Collectors.toSet());
 
             // Bail out if we have all the entries requested
             if (candidateMissingExternalKeys.isEmpty())
             {
                 getLogger().trace("Cache {}: getBulk(Function): had all the requested entries cached", name);
                 return grandResult;
             }
             getLogger().trace("Cache {}: getBulk(Function): checking external cache for {} keys",
                     name, candidateMissingExternalKeys.size());
 
             final Map<String, V> missingValues = handleCreation(factory, candidateMissingExternalKeys);
             cacheContext.recordValues(missingValues);
             grandResult.putAll(missingValues);
 
             return grandResult;
         });
     }
 }