package com.atlassian.vcache.internal.guava;
   
   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 com.google.common.cache.Cache;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.time.Duration;
  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;
  
  /**
   * Guava implementation of a {@link com.atlassian.vcache.TransactionalExternalCache}.
   *
   * @param <V> the value type.
   * @since 1.0.0
   */
  public class GuavaTransactionalExternalCache<V>
          extends AbstractTransactionalExternalCache<V> {
      private static final Logger log = LoggerFactory.getLogger(GuavaTransactionalExternalCache.class);
  
      private final Cache<String, IdentifiedData> delegate;
      private final ExternalCacheKeyGenerator keyGenerator;
      private final Optional<MarshallingPair<V>> valueMarshalling;
      private final TransactionControlManager transactionControlManager;
  
      public GuavaTransactionalExternalCache(
              String name,
              Cache<String, IdentifiedData> delegate,
              Supplier<RequestContext> contextSupplier,
              ExternalCacheKeyGenerator keyGenerator,
              Optional<MarshallingPair<V>> valueMarshalling,
              TransactionControlManager transactionControlManager,
              MetricsRecorder metricsRecorder,
              Duration lockTimeout) {
          super(name, contextSupplier, metricsRecorder, lockTimeout, (n, ex) -> {});
          this.delegate = requireNonNull(delegate);
          this.keyGenerator = requireNonNull(keyGenerator);
          this.valueMarshalling = requireNonNull(valueMarshalling);
          this.transactionControlManager = requireNonNull(transactionControlManager);
      }
  
      @Override
      public void transactionSync() {
          log.trace("Cache {}: synchronising operations", name);
          final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
  
          if (cacheContext.hasRemoveAll()) {
              delegate.asMap().clear();
          }
  
          performKeyedOperations(cacheContext);
          cacheContext.forgetAll();
      }
  
      @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, getName(), requestContext::partitionIdentifier, lockTimeout);
          });
      }
  
      @Override
      protected Logger getLogger() {
          return log;
      }
  
      @Override
      protected final ExternalCacheException mapException(Exception ex) {
          return GuavaUtils.mapException(ex);
      }
  
      @Override
      protected final Optional<V> directGet(String externalKey) {
          return unmarshall(delegate.getIfPresent(externalKey), valueMarshalling);
      }
 
     @Override
     protected final Map<String, Optional<V>> directGetBulk(Set<String> externalKeys) {
         return GuavaUtils.directGetBulk(externalKeys, delegate, valueMarshalling);
     }
 
     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.asMap().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 =
                             GuavaUtils.directPut(
                                     externalKey,
                                     identifiedData,
                                     entry.getValue().getPolicy(),
                                     delegate);
 
                     if (!putOutcome) {
                         log.debug("Cache {}: Unable to perform put() operation {} on entry {}",
                                 name, entry.getValue().getPolicy(), entry.getKey());
 
                         delegate.asMap().clear();
                         break;
                     }
                 }
             }
         } catch (ExternalCacheException bugger) {
             log.error("Cache {}: an operation failed in transaction sync, so clearing the cache", name);
             delegate.asMap().clear();
         }
     }
 
 }