package com.atlassian.vcache.internal.legacy;
   
   import com.atlassian.cache.Cache;
   import com.atlassian.cache.CacheException;
   import com.atlassian.marshalling.api.MarshallingPair;
   import com.atlassian.vcache.ExternalCacheException;
   import com.atlassian.vcache.internal.RequestContext;
   import com.atlassian.vcache.internal.core.ExternalCacheKeyGenerator;
   import com.atlassian.vcache.internal.core.TransactionControlManager;
  import com.atlassian.vcache.internal.core.cas.IdentifiedData;
  import com.atlassian.vcache.internal.core.metrics.MetricsRecorder;
  import com.atlassian.vcache.internal.core.service.AbstractExternalCacheRequestContext;
  import com.atlassian.vcache.internal.core.service.AbstractTransactionalExternalCache;
  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.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 backed on to Atlassian Cache.
   *
   * @param <V> the value type
   * @since 1.0.0
   */
  class LegacyTransactionalExternalCache<V>
          extends AbstractTransactionalExternalCache<V> {
      private static final Logger log = LoggerFactory.getLogger(LegacyTransactionalExternalCache.class);
  
      private final Cache<String, IdentifiedData> delegate;
      private final ExternalCacheKeyGenerator keyGenerator;
      private final Optional<MarshallingPair<V>> valueMarshalling;
      private final TransactionControlManager transactionControlManager;
      private final LegacyServiceSettings serviceSettings;
  
      LegacyTransactionalExternalCache(
              Cache<String, IdentifiedData> delegate,
              Supplier<RequestContext> contextSupplier,
              ExternalCacheKeyGenerator keyGenerator,
              Optional<MarshallingPair<V>> valueMarshalling,
              TransactionControlManager transactionControlManager,
              LegacyServiceSettings serviceSettings,
              MetricsRecorder metricsRecorder) {
          super(delegate.getName(),
                  contextSupplier,
                  metricsRecorder,
                  serviceSettings.getLockTimeout(),
                  (n, e) -> {});
          this.delegate = requireNonNull(delegate);
          this.keyGenerator = requireNonNull(keyGenerator);
          this.valueMarshalling = requireNonNull(valueMarshalling);
          this.transactionControlManager = requireNonNull(transactionControlManager);
          this.serviceSettings = requireNonNull(serviceSettings);
      }
  
      @Override
      public void transactionSync() {
          log.trace("Cache {}: synchronising operations", name);
          final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
  
          if (cacheContext.hasRemoveAll()) {
              delegate.removeAll();
          }
  
          performKeyedOperations(cacheContext);
          cacheContext.forgetAll();
      }
  
      private void performKeyedOperations(AbstractExternalCacheRequestContext<V> cacheContext) {
          try {
              for (Map.Entry<String, AbstractExternalCacheRequestContext.DeferredOperation<V>> entry
                      : cacheContext.getKeyedOperations()) {
                  final String externalKey = cacheContext.externalEntryKeyFor(entry.getKey());
  
                  if (entry.getValue().isRemove()) {
                      log.trace("Cache {}: performing remove on entry {}", name, entry.getKey());
                      delegate.remove(externalKey);
                  } else {
                      log.trace("Cache {}: performing {} on entry {}", name, entry.getValue().getPolicy(), entry.getKey());
                      final IdentifiedData identifiedData = marshall(entry.getValue().getValue(), valueMarshalling);
  
                      final boolean putOutcome =
                              LegacyUtils.directPut(
                                      externalKey,
                                      identifiedData,
                                      entry.getValue().getPolicy(),
                                      delegate,
                                      serviceSettings.isAvoidCasOps());
  
                      if (!putOutcome) {
                          log.debug("Cache {}: Unable to perform put() operation {} on entry {}, clearing cache",
                                  name, entry.getValue().getPolicy(), entry.getKey());
                         delegate.removeAll();
                         break;
                     }
                 }
             }
         } catch (ExternalCacheException | CacheException bugger) {
             log.error("Cache {}: an operation failed in transaction sync, so clearing the cache", name, bugger);
             delegate.removeAll();
         }
     }
 
     @Override
     protected AbstractExternalCacheRequestContext<V> ensureCacheContext() {
         final RequestContext requestContext = contextSupplier.get();
 
         transactionControlManager.registerTransactionalExternalCache(requestContext, name, this);
 
         return requestContext.computeIfAbsent(this, () -> {
             // Need to build a new context, which involves getting the current cache version, or setting it if it does
             // not exist.
             log.trace("Cache {}: Setting up a new context", name);
             return new UnversionedExternalCacheRequestContext<>(
                     keyGenerator,
                     delegate.getName(),
                     requestContext::partitionIdentifier,
                     serviceSettings.getLockTimeout());
         });
     }
 
     @Override
     protected Logger getLogger() {
         return log;
     }
 
     @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);
     }
 }