package com.atlassian.vcache.internal.memcached;
   
   import com.atlassian.marshalling.jdk.StringMarshalling;
   import com.atlassian.vcache.DirectExternalCache;
   import com.atlassian.vcache.ExternalCacheSettings;
   import com.atlassian.vcache.StableReadExternalCache;
   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.Sha1ExternalCacheKeyGenerator;
  import com.atlassian.vcache.internal.core.metrics.DefaultMetricsCollector;
  import com.atlassian.vcache.internal.core.metrics.MetricsCollector;
  import com.atlassian.vcache.internal.test.AbstractUnreliableStableReadExternalCacheIT;
  import net.spy.memcached.AddrUtil;
  import net.spy.memcached.BinaryConnectionFactory;
  import net.spy.memcached.MemcachedClient;
  import org.junit.BeforeClass;
  
  import java.io.IOException;
  import java.time.Duration;
  
  public class MemcachedStableReadExternalCacheIT extends AbstractUnreliableStableReadExternalCacheIT {
      private static MemcachedClient client;
      private static MemcachedClient offlineClient; // Points to a non-responsive memcached
      private RequestContext requestContext = new DefaultRequestContext(() -> "tenant-id");
      private final MetricsCollector metricsCollector = new DefaultMetricsCollector(() -> 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")); // Unassigned port
      }
  
      @Override
      protected StableReadExternalCache<String> createCache(String name, ExternalCacheSettings settings, Duration lockTimeout) {
          return createCache(name, settings, lockTimeout, client);
      }
  
      private StableReadExternalCache<String> createCache(String name, ExternalCacheSettings settings, Duration lockTimeout, MemcachedClient client) {
          return metricsCollector.wrap(
                  new MemcachedStableReadExternalCache<>(
                          Utils.defaultServiceSettingsBuilder(() -> client, lockTimeout).build(),
                          () -> requestContext,
                          new Sha1ExternalCacheKeyGenerator("it-test"),
                          name,
                          StringMarshalling.pair(),
                          settings,
                          metricsCollector));
      }
  
      @Override
      protected StableReadExternalCache<String> createOfflineCache(String name, ExternalCacheSettings settings, Duration lockTimeout) {
          return createCache(name, settings, lockTimeout, offlineClient);
      }
  
      @Override
      protected DirectExternalCache<String> obtainDirectCache(String name, ExternalCacheSettings settings) {
          return new MemcachedDirectExternalCache<>(
                  Utils.defaultServiceSettingsBuilder(() -> client, Duration.ofSeconds(1)).build(),
                  () -> requestContext,
                  new Sha1ExternalCacheKeyGenerator("it-test"),
                  name,
                  StringMarshalling.pair(),
                  settings);
      }
  
      @Override
      protected RequestMetrics requestMetrics() {
          return metricsCollector.obtainRequestMetrics(requestContext);
      }
  }