package com.atlassian.vcache.internal.memcached;
   
   import com.atlassian.marshalling.api.MarshallingException;
   import com.atlassian.marshalling.api.MarshallingPair;
   import com.atlassian.vcache.ExternalCacheException;
   import com.atlassian.vcache.ExternalCacheSettings;
   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.VCacheCoreUtils;
  import com.atlassian.vcache.internal.core.metrics.MetricsRecorder;
  import com.atlassian.vcache.internal.core.service.AbstractExternalCacheRequestContext.DeferredOperation;
  import com.atlassian.vcache.internal.core.service.AbstractTransactionalExternalCache;
  import com.atlassian.vcache.internal.core.service.VersionedExternalCacheRequestContext;
  import net.spy.memcached.MemcachedClientIF;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.util.HashMap;
  import java.util.Map;
  import java.util.Map.Entry;
  import java.util.Optional;
  import java.util.Set;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.Future;
  import java.util.function.Supplier;
  
  import static com.atlassian.vcache.ExternalCacheException.Reason.TRANSACTION_FAILURE;
  import static com.atlassian.vcache.VCacheUtils.fold;
  import static com.atlassian.vcache.internal.core.VCacheCoreUtils.unmarshall;
  import static com.atlassian.vcache.internal.memcached.MemcachedUtils.expiryTime;
  import static com.atlassian.vcache.internal.memcached.MemcachedUtils.putOperationForPolicy;
  import static java.util.Objects.requireNonNull;
  
  /**
   * Implementation that backs onto Memcached.
   *
   * @param <V> the value type
   * @since 1.0.0
   */
  class MemcachedTransactionalExternalCache<V>
          extends AbstractTransactionalExternalCache<V> {
      private static final Logger log = LoggerFactory.getLogger(MemcachedTransactionalExternalCache.class);
  
      private final Supplier<MemcachedClientIF> clientSupplier;
      private final ExternalCacheKeyGenerator keyGenerator;
      private final MarshallingPair<V> valueMarshalling;
      private final int ttlSeconds;
      private final TransactionControlManager transactionControlManager;
  
      MemcachedTransactionalExternalCache(
              MemcachedVCacheServiceSettings serviceSettings,
              Supplier<RequestContext> contextSupplier,
              ExternalCacheKeyGenerator keyGenerator,
              String name,
              MarshallingPair<V> valueMarshalling,
              ExternalCacheSettings settings,
              TransactionControlManager transactionControlManager,
              MetricsRecorder metricsRecorder) {
          super(name, contextSupplier, metricsRecorder, serviceSettings.getLockTimeout(), serviceSettings.getExternalCacheExceptionListener());
          this.clientSupplier = requireNonNull(serviceSettings.getClientSupplier());
          this.keyGenerator = requireNonNull(keyGenerator);
          this.valueMarshalling = requireNonNull(valueMarshalling);
          this.ttlSeconds = VCacheCoreUtils.roundUpToSeconds(settings.getDefaultTtl().get());
          this.transactionControlManager = requireNonNull(transactionControlManager);
      }
  
      @Override
      public void transactionSync() {
          log.trace("Cache {}: synchronising operations", name);
          final VersionedExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
  
          fold(perform(() -> {
                      if (cacheContext.hasRemoveAll()) {
                          cacheContext.updateCacheVersion(MemcachedUtils.cacheVersionIncrementer(clientSupplier));
                      }
  
                      performKeyedOperations(cacheContext);
  
                      return null;
                  }),
                  v -> null,
                  err -> {
                      log.warn("Cache {}: an operation failed during transaction sync ({}). Clearing cache to remove stale entries.", name, err.getMessage());
                      try {
                          cacheContext.updateCacheVersion(MemcachedUtils.cacheVersionIncrementer(clientSupplier));
                      } catch (RuntimeException e) {
                          log.error("Cache {}: failed to clear the cache: {}", name, e.getMessage());
                      }
  
                      return null;
                  });
  
          cacheContext.forgetAll();
      }
  
      private void performKeyedOperations(VersionedExternalCacheRequestContext<V> cacheContext) {
          // The approach used to perform the operations is:
          // - schedule all the operations
         // - check the result of all the operations and log a warning is necessary
         // - if any operation failed, wipe the cache
 
         // Right then, lets submit all the operations.
         final Map<Future<Boolean>, String> futureToFailureMessageMap = new HashMap<>();
         final boolean[] anOperationFailed = new boolean[1];
 
         for (Entry<String, 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());
                 futureToFailureMessageMap.put(
                         clientSupplier.get().delete(externalKey),
                         "remove entry " + entry.getKey());
             } else {
                 log.trace("Cache {}: performing {} on entry {}", name, entry.getValue().getPolicy(), entry.getKey());
                 try {
                     final byte[] valueBytes = valueMarshalling.getMarshaller().marshallToBytes(
                             requireNonNull(entry.getValue().getValue()));
                     final Future<Boolean> putOp = putOperationForPolicy(
                             entry.getValue().getPolicy(),
                             clientSupplier.get(),
                             externalKey,
                             expiryTime(ttlSeconds),
                             valueBytes);
                     futureToFailureMessageMap.put(
                             putOp,
                             "put using policy " + entry.getValue().getPolicy() + " on entry "
                                     + entry.getKey());
                 } catch (MarshallingException mex) {
                     log.warn("Cache {}: Unable to marshall value to perform put operation on entry {}",
                             name, entry.getKey(), mex);
                     anOperationFailed[0] = true;
                     break;
                 }
             }
         }
 
         // Cool, now lets check the result of each operation
         futureToFailureMessageMap.entrySet().forEach(entry -> {
             try {
                 if (anOperationFailed[0]) {
                     // We can safely cancel while running - see javadoc for net.spy.memcached.MemcachedClient
                     entry.getKey().cancel(true);
                 } else {
                     final Boolean outcome = entry.getKey().get();
                     if (outcome) {
                         log.trace("Cache {}: successful deferred operation for {}", name, entry.getValue());
                     } else {
                         anOperationFailed[0] = true;
                         log.warn("Cache {}: failed deferred operation for {}", name, entry.getValue());
                     }
                 }
             } catch (InterruptedException | ExecutionException ex) {
                 if (ex instanceof InterruptedException) {
                     // re-interrupt the thread if we've swallowed an interrupt - https://www.ibm.com/developerworks/library/j-jtp05236/
                     Thread.currentThread().interrupt();
                 }
                 anOperationFailed[0] = true;
                 log.error("Cache {}: had failure getting result for deferred operation {}",
                         name, entry.getValue(), ex);
             }
         });
 
         // Finally, lets see if an operation failed
         if (anOperationFailed[0]) {
             throw new ExternalCacheException(TRANSACTION_FAILURE);
         }
     }
 
     @Override
     protected VersionedExternalCacheRequestContext<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 VersionedExternalCacheRequestContext<>(
                     keyGenerator, name, requestContext::partitionIdentifier,
                     MemcachedUtils.cacheVersionSupplier(clientSupplier),
                     lockTimeout);
         });
     }
 
     @Override
     protected Logger getLogger() {
         return log;
     }
 
     @Override
     protected final ExternalCacheException mapException(Exception ex) {
         return MemcachedUtils.mapException(ex);
     }
 
     @Override
     protected final Optional<V> directGet(String externalKey) {
         return unmarshall((byte[]) clientSupplier.get().get(externalKey), valueMarshalling);
     }
 
     @Override
     protected final Map<String, Optional<V>> directGetBulk(Set<String> externalKeys) {
         return MemcachedUtils.directGetBulk(externalKeys, clientSupplier, valueMarshalling);
     }
 }