package com.atlassian.vcache.internal.legacy;
   
   import com.atlassian.cache.Cache;
   import com.atlassian.marshalling.api.MarshallingPair;
   import com.atlassian.vcache.ExternalCacheException;
   import com.atlassian.vcache.PutPolicy;
   import com.atlassian.vcache.internal.MetricLabel;
   import com.atlassian.vcache.internal.RequestContext;
   import com.atlassian.vcache.internal.core.ExternalCacheKeyGenerator;
  import com.atlassian.vcache.internal.core.cas.IdentifiedData;
  import com.atlassian.vcache.internal.core.metrics.CacheType;
  import com.atlassian.vcache.internal.core.metrics.MetricsRecorder;
  import com.atlassian.vcache.internal.core.service.AbstractExternalCacheRequestContext;
  import com.atlassian.vcache.internal.core.service.AbstractStableReadExternalCache;
  import com.atlassian.vcache.internal.core.service.UnversionedExternalCacheRequestContext;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.util.Map;
  import java.util.Optional;
  import java.util.Set;
  import java.util.concurrent.ExecutionException;
  import java.util.function.Supplier;
  
  import static com.atlassian.vcache.internal.core.cas.IdentifiedUtils.marshall;
  import static com.atlassian.vcache.internal.core.cas.IdentifiedUtils.unmarshall;
  import static java.util.Objects.requireNonNull;
  
  /**
   * Implementation that backs onto Atlassian Cache.
   *
   * @param <V> the value type
   * @since 1.0.0
   */
  class LegacyStableReadExternalCache<V>
          extends AbstractStableReadExternalCache<V> {
      private static final Logger log = LoggerFactory.getLogger(LegacyStableReadExternalCache.class);
  
      private final Cache<String, IdentifiedData> delegate;
      private final Supplier<RequestContext> contextSupplier;
      private final ExternalCacheKeyGenerator keyGenerator;
      private final Optional<MarshallingPair<V>> valueMarshalling;
      private final LegacyServiceSettings serviceSettings;
  
      LegacyStableReadExternalCache(
              Cache<String, IdentifiedData> delegate,
              Supplier<RequestContext> contextSupplier,
              ExternalCacheKeyGenerator keyGenerator,
              Optional<MarshallingPair<V>> valueMarshalling,
              LegacyServiceSettings serviceSettings,
              MetricsRecorder metricsRecorder) {
          super(delegate.getName(), metricsRecorder, serviceSettings.getLockTimeout());
          this.delegate = requireNonNull(delegate);
          this.contextSupplier = requireNonNull(contextSupplier);
          this.keyGenerator = requireNonNull(keyGenerator);
          this.valueMarshalling = requireNonNull(valueMarshalling);
          this.serviceSettings = requireNonNull(serviceSettings);
      }
  
      @Override
      public boolean internalPut(String internalKey, V value, PutPolicy policy) {
          final String externalKey = ensureCacheContext().externalEntryKeyFor(internalKey);
          final IdentifiedData identifiedData = marshall(value, valueMarshalling);
          return LegacyUtils.directPut(
                  externalKey, identifiedData, policy, delegate, serviceSettings.isAvoidCasOps());
      }
  
      @Override
      protected void internalRemove(Iterable<String> internalKeys) {
          // There is no bulk delete in the api, so need to remove each one
          final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
          for (String key : internalKeys) {
              delegate.remove(cacheContext.externalEntryKeyFor(key));
              cacheContext.recordValue(key, Optional.empty());
          }
      }
  
      @Override
      protected void internalRemoveAll() {
          delegate.removeAll();
      }
  
      @Override
      protected Logger getLogger() {
          return log;
      }
  
      @Override
      protected AbstractExternalCacheRequestContext<V> ensureCacheContext() {
          final RequestContext requestContext = contextSupplier.get();
  
          return requestContext.computeIfAbsent(this, () -> {
              log.trace("Cache {}: Setting up a new context", delegate.getName());
              return new UnversionedExternalCacheRequestContext<>(
                      keyGenerator, delegate.getName(), requestContext::partitionIdentifier, lockTimeout);
          });
      }
  
      @Override
     protected V handleCreation(String internalKey, V candidateValue)
             throws ExecutionException, InterruptedException {
         final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
         final IdentifiedData candidateIdentifiedData = marshall(candidateValue, valueMarshalling);
         final String externalKey = cacheContext.externalEntryKeyFor(internalKey);
 
         metricsRecorder.record(name, CacheType.EXTERNAL, MetricLabel.NUMBER_OF_REMOTE_GET, 1);
         if (serviceSettings.isAvoidCasOps()) {
             delegate.put(externalKey, candidateIdentifiedData);
         } else {
             final Optional<V> otherAddedValue =
                     unmarshall(delegate.putIfAbsent(externalKey, candidateIdentifiedData), valueMarshalling);
 
             if (otherAddedValue.isPresent()) {
                 getLogger().info("Cache {}, unable to add candidate for key {}, use what was added", name, internalKey);
                 // Record as if doing another remote call
                 metricsRecorder.record(name, CacheType.EXTERNAL, MetricLabel.NUMBER_OF_REMOTE_GET, 1);
                 return otherAddedValue.get();
             }
         }
         return candidateValue;
     }
 
     @Override
     protected final ExternalCacheException mapException(Exception ex) {
         return LegacyUtils.mapException(ex);
     }
 
     @Override
     protected final Optional<V> directGet(String externalKey) {
         return unmarshall(delegate.get(externalKey), valueMarshalling);
     }
 
     @Override
     protected final Map<String, Optional<V>> directGetBulk(Set<String> externalKeys) {
         return LegacyUtils.directGetBulk(externalKeys, delegate, valueMarshalling);
     }
 }