package com.atlassian.vcache.internal.memcached;
   
   import com.atlassian.marshalling.jdk.StringMarshalling;
   import com.atlassian.vcache.ChangeRate;
   import com.atlassian.vcache.DirectExternalCache;
   import com.atlassian.vcache.ExternalCacheSettings;
   import com.atlassian.vcache.ExternalCacheSettingsBuilder;
   import com.atlassian.vcache.TransactionalExternalCache;
   import com.atlassian.vcache.internal.RequestContext;
  import com.atlassian.vcache.internal.RequestMetrics;
  import com.atlassian.vcache.internal.core.DefaultRequestContext;
  import com.atlassian.vcache.internal.core.DefaultTransactionControlManager;
  import com.atlassian.vcache.internal.core.Sha1ExternalCacheKeyGenerator;
  import com.atlassian.vcache.internal.core.metrics.DefaultMetricsCollector;
  import com.atlassian.vcache.internal.core.metrics.MetricsCollector;
  import com.atlassian.vcache.internal.test.AbstractUnreliableTransactionalExternalCacheIT;
  import net.spy.memcached.AddrUtil;
  import net.spy.memcached.BinaryConnectionFactory;
  import net.spy.memcached.MemcachedClient;
  import org.junit.Before;
  import org.junit.BeforeClass;
  
  import java.io.IOException;
  import java.time.Duration;
  import java.util.concurrent.CompletionStage;
  
  import static com.atlassian.vcache.internal.test.CompletionStageSuccessful.successful;
  import static org.hamcrest.MatcherAssert.assertThat;
  
  public class MemcachedTransactionalExternalCacheIT extends AbstractUnreliableTransactionalExternalCacheIT {
      private static MemcachedClient client;
      private static MemcachedClient offlineClient; // Points to a non-responsive memcached
  
      private final RequestContext requestContext = new DefaultRequestContext(() -> "txn-it-test");
      private final MetricsCollector metricsCollector = new DefaultMetricsCollector(() -> requestContext);
  
      private MemcachedTransactionalExternalCache<String> cache;
      private TransactionalExternalCache<String> wrappedCache;
      private MemcachedDirectExternalCache<String> directCache;
  
      private MemcachedTransactionalExternalCache<String> offlineCache; // Points to a non-responsive memcached
      private MemcachedDirectExternalCache<String> offlineDirectCache;
  
      @Override
      protected TransactionalExternalCache<String> cache() {
          return wrappedCache;
      }
  
      @Override
      protected DirectExternalCache<String> directCache() {
          return directCache;
      }
  
      @Override
      protected TransactionalExternalCache<String> offlineCache() {
          return offlineCache;
      }
  
      @Override
      protected DirectExternalCache<String> offlineDirectCache() {
          return offlineDirectCache;
      }
  
      @Override
      protected void directCacheRefresh() {
          directCache.refreshCacheVersion();
      }
  
      @Override
      protected void cacheTransactionSync() {
          cache.transactionSync();
      }
  
      @Override
      protected void cacheTransactionDiscard() {
          cache.transactionDiscard();
      }
  
      @Override
      protected void offlineCacheTransactionSync() {
          offlineCache.transactionSync();
      }
  
      @Override
      protected void offlineCacheTransactionDiscard() {
          offlineCache.transactionDiscard();
      }
  
      @Override
      protected RequestMetrics requestMetrics() {
          return metricsCollector.obtainRequestMetrics(requestContext);
      }
  
      @BeforeClass
      public static void init_client() throws IOException {
          client = new MemcachedClient(
                  new BinaryConnectionFactory(),
                  AddrUtil.getAddresses(System.getProperty("memcached.svr")));
  
         offlineClient = new MemcachedClient(
                 new BinaryConnectionFactory(),
                 AddrUtil.getAddresses("localhost:4"));
     }
 
     @Before
     public void ensureCache() {
         final ExternalCacheSettings settings = new ExternalCacheSettingsBuilder()
                 .entryGrowthRateHint(ChangeRate.LOW_CHANGE)
                 .entryCountHint(5)
                 .defaultTtl(Duration.ofSeconds(60))
                 .dataChangeRateHint(ChangeRate.HIGH_CHANGE)
                 .build();
         final MemcachedVCacheServiceSettings serviceSettings =
                 Utils.defaultServiceSettingsBuilder(() -> client, Duration.ofSeconds(1)).build();
         cache = new MemcachedTransactionalExternalCache<>(
                 serviceSettings,
                 () -> requestContext,
                 new Sha1ExternalCacheKeyGenerator("it-test"),
                 CACHE_NAME,
                 StringMarshalling.pair(),
                 settings,
                 new DefaultTransactionControlManager(metricsCollector, context -> {}),
                 metricsCollector);
         wrappedCache = metricsCollector.wrap(cache);
 
         directCache = new MemcachedDirectExternalCache<>(
                 serviceSettings,
                 () -> requestContext,
                 new Sha1ExternalCacheKeyGenerator("it-test"),
                 CACHE_NAME,
                 StringMarshalling.pair(),
                 settings);
 
         // Start from a clean slate
         final CompletionStage<Void> rm = directCache.removeAll();
         assertThat(rm, successful());
 
         offlineCache = new MemcachedTransactionalExternalCache<>(
                 Utils.defaultServiceSettingsBuilder(() -> offlineClient, Duration.ofSeconds(1)).build(),
                 () -> requestContext,
                 new Sha1ExternalCacheKeyGenerator("it-test-offline"),
                 CACHE_NAME + ".offline",
                 StringMarshalling.pair(),
                 settings,
                 new DefaultTransactionControlManager(metricsCollector, context -> {}),
                 metricsCollector);
 
         offlineDirectCache = new MemcachedDirectExternalCache<>(
                 serviceSettings,
                 () -> requestContext,
                 new Sha1ExternalCacheKeyGenerator("it-test-offline"),
                 CACHE_NAME + ".offline",
                 StringMarshalling.pair(),
                 settings);
     }
 }