Merge pull request #56 from artkoshelev/make-polling-strategy-customizable

ISSUE-55 make config polling customizable

Author: YueTang-Vanessa

src/main/java/net/spy/memcached/ConfigurationPoller.java

@@ -7,10 +7,8 @@
 
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
-import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -20,6 +18,8 @@
 import net.spy.memcached.compat.SpyThread;
 import net.spy.memcached.config.ClusterConfiguration;
 import net.spy.memcached.config.ClusterConfigurationObserver;
+import net.spy.memcached.config.ConfigEndpointSelectionStrategy;
+import net.spy.memcached.config.RoundRobinConfigEndpointSelectionStrategy;
 import net.spy.memcached.config.NodeEndPoint;
 import net.spy.memcached.ops.ConfigurationType;
 import net.spy.memcached.transcoders.SerializingTranscoder;
@@ -46,21 +46,23 @@ public class ConfigurationPoller extends SpyThread{
   private ClusterConfiguration currentClusterConfiguration;
   //The transcoder used for config.
   private Transcoder<Object> configTranscoder = new SerializingTranscoder();
-  private int currentIndex = 0;
   private Date date = new Date();
   private long lastSuccessfulPoll = date.getTime(); 
   private int pollingErrorCount = 0;
+  private final ConfigEndpointSelectionStrategy configEndpointSelectionStrategy;
 
   //The executor is used to keep the task and it's execution independent. The scheduled thread polls takes care of 
   //the periodic polling.
   private ScheduledThreadPoolExecutor scheduledExecutor;
 
   public ConfigurationPoller(final MemcachedClient client){
-    this(client, DEFAULT_POLL_INTERVAL, false);
+    this(client, DEFAULT_POLL_INTERVAL, false, new RoundRobinConfigEndpointSelectionStrategy());
   }
 
-  public ConfigurationPoller(final MemcachedClient client, long pollingInterval, final boolean useDaemonThreads){
+  public ConfigurationPoller(final MemcachedClient client, long pollingInterval, final boolean useDaemonThreads,
+                             final ConfigEndpointSelectionStrategy configEndpointSelectionStrategy){
     this.client = client;
+    this.configEndpointSelectionStrategy = configEndpointSelectionStrategy;
     this.scheduledExecutor = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
 		@Override
 		public Thread newThread(Runnable runnable) {
@@ -90,21 +92,8 @@ public void run(){
     try{
       getLogger().info("Starting configuration poller.");
       String newConfigResponse = null;
-      NodeEndPoint endpointToGetConfig = null;
-    
-      Collection<NodeEndPoint> endpoints = client.getAvailableNodeEndPoints();
-      if(endpoints.isEmpty()){
-        //If no nodes are available status, then get all the endpoints. This provides an 
-        //oppurtunity to re-resolve the hostname by recreating InetSocketAddress instance in "NodeEndPoint".getInetSocketAddress().
-        endpoints = client.getAllNodeEndPoints();
-      }
-      currentIndex = (currentIndex+1)%endpoints.size();
-      Iterator<NodeEndPoint> iterator = endpoints.iterator();
-      for(int i =0;i<currentIndex;i++){
-        iterator.next();
-      }
+      NodeEndPoint endpointToGetConfig = configEndpointSelectionStrategy.getConfigEndpoint(client);
 
-      endpointToGetConfig = iterator.next();
       InetSocketAddress socketAddressToGetConfig = endpointToGetConfig.getInetSocketAddress();
 
       getLogger().info("Endpoint to use for configuration access in this poll " + endpointToGetConfig.toString());

src/main/java/net/spy/memcached/ConnectionFactory.java

@@ -38,6 +38,7 @@
 import java.util.concurrent.ExecutorService;
 
 import net.spy.memcached.auth.AuthDescriptor;
+import net.spy.memcached.config.ConfigEndpointSelectionStrategy;
 import net.spy.memcached.metrics.MetricCollector;
 import net.spy.memcached.metrics.MetricType;
 import net.spy.memcached.ops.Operation;
@@ -128,7 +129,13 @@ MemcachedNode createMemcachedNode(SocketAddress sa, SocketChannel c,
    * @return the interval in milliseconds
    */
   long getDynamicModePollingInterval();
-  
+
+  /**
+   * The strategy used for selecting which endpoint to use when polling for cluster configuration.
+   * @return the configuration endpoint selection strategy
+   */
+  ConfigEndpointSelectionStrategy getConfigEndpointSelectionStrategy();
+
   /**
    * Get the operation timeout used by this connection.
    */

src/main/java/net/spy/memcached/ConnectionFactoryBuilder.java

@@ -35,6 +35,7 @@
 import java.util.concurrent.ExecutorService;
 
 import net.spy.memcached.auth.AuthDescriptor;
+import net.spy.memcached.config.ConfigEndpointSelectionStrategy;
 import net.spy.memcached.metrics.MetricCollector;
 import net.spy.memcached.metrics.MetricType;
 import net.spy.memcached.ops.Operation;
@@ -57,7 +58,9 @@ public class ConnectionFactoryBuilder {
   protected Transcoder<Object> transcoder;
 
   protected ClientMode clientMode;
-  
+
+  protected ConfigEndpointSelectionStrategy configEndpointSelectionStrategy;
+
   protected FailureMode failureMode;
 
   protected Collection<ConnectionObserver> initialObservers =
@@ -121,6 +124,7 @@ public ConnectionFactoryBuilder(ConnectionFactory cf) {
     setSSLContext(cf.getSSLContext());
     setHostnameForTlsVerification(cf.getHostnameForTlsVerification());
     setSkipTlsHostnameVerification(cf.skipTlsHostnameVerification());
+    setConfigEndpointSelectionStrategy(cf.getConfigEndpointSelectionStrategy());
   }
 
   /**
@@ -133,6 +137,11 @@ public ConnectionFactoryBuilder setClientMode(ClientMode clientMode){
     return this;
   }
 
+  public ConnectionFactoryBuilder setConfigEndpointSelectionStrategy(ConfigEndpointSelectionStrategy configEndpointSelectionStrategy){
+    this.configEndpointSelectionStrategy = configEndpointSelectionStrategy;
+    return this;
+  }
+
   public ConnectionFactoryBuilder setOpQueueFactory(OperationQueueFactory q) {
     opQueueFactory = q;
     return this;
@@ -388,6 +397,13 @@ public ConnectionFactory build() {
       public ClientMode getClientMode(){
         return clientMode == null ? super.getClientMode() : clientMode;
       }
+
+      @Override
+      public ConfigEndpointSelectionStrategy getConfigEndpointSelectionStrategy() {
+        return configEndpointSelectionStrategy == null
+          ? super.getConfigEndpointSelectionStrategy()
+          : configEndpointSelectionStrategy;
+      }
 
       @Override
       public BlockingQueue<Operation> createOperationQueue() {

src/main/java/net/spy/memcached/DefaultConnectionFactory.java

@@ -42,6 +42,8 @@
 
 import net.spy.memcached.auth.AuthDescriptor;
 import net.spy.memcached.compat.SpyObject;
+import net.spy.memcached.config.ConfigEndpointSelectionStrategy;
+import net.spy.memcached.config.RoundRobinConfigEndpointSelectionStrategy;
 import net.spy.memcached.metrics.DefaultMetricCollector;
 import net.spy.memcached.metrics.MetricCollector;
 import net.spy.memcached.metrics.MetricType;
@@ -136,6 +138,7 @@ public class DefaultConnectionFactory extends SpyObject implements
   protected final int opQueueLen;
   private final int readBufSize;
   private final HashAlgorithm hashAlg;
+  private final ConfigEndpointSelectionStrategy configEndpointSelectionStrategy;
 
   private MetricCollector metrics;
 
@@ -162,6 +165,7 @@ public DefaultConnectionFactory(ClientMode clientMode, int qLen, int bufSize, Ha
     this.readBufSize = bufSize;
     this.hashAlg = hash;
     this.metrics = null;
+    this.configEndpointSelectionStrategy = new RoundRobinConfigEndpointSelectionStrategy();
   }
 
   /**
@@ -215,6 +219,10 @@ public long getDynamicModePollingInterval(){
     return ConfigurationPoller.DEFAULT_POLL_INTERVAL;
   }
 
+  public ConfigEndpointSelectionStrategy getConfigEndpointSelectionStrategy() {
+    return configEndpointSelectionStrategy;
+  }
+
   public MemcachedNode createMemcachedNode(SocketAddress sa, SocketChannel c,
       int bufSize) {
 

src/main/java/net/spy/memcached/MemcachedClient.java

@@ -33,6 +33,7 @@
 import net.spy.memcached.compat.SpyObject;
 import net.spy.memcached.config.ClusterConfiguration;
 import net.spy.memcached.ConfigurationPoller;
+import net.spy.memcached.config.ConfigEndpointSelectionStrategy;
 import net.spy.memcached.config.NodeEndPoint;
 import net.spy.memcached.internal.BulkFuture;
 import net.spy.memcached.internal.BulkGetFuture;
@@ -191,12 +192,14 @@ public class MemcachedClient extends SpyObject implements MemcachedClientIF,
 
   protected final ExecutorService executorService;
 
+  protected final ConfigEndpointSelectionStrategy configEndpointSelectionStrategy;
+
   private NodeEndPoint configurationNode;
   //Set default value to true to attempt config API first. The value is set to false if
   //OperationNotSupportedException is thrown.
   private boolean isConfigurationProtocolSupported = true;
 
-  //This is used to dynamic mode to track whether the client is initialized with set of cache nodes for the first time.
+  // This is used in dynamic mode to track whether the client is initialized with a set of cache nodes for the first time.
   private boolean isConfigurationInitialized = false;
 
   private Transcoder<Object> configTranscoder = new SerializingTranscoder();
@@ -290,6 +293,8 @@ private MemcachedClient(ConnectionFactory cf, List<InetSocketAddress> addrs, boo
     authDescriptor = cf.getAuthDescriptor();
     executorService = cf.getListenerExecutorService();
 
+    configEndpointSelectionStrategy = cf.getConfigEndpointSelectionStrategy();
+
     if(clientMode == ClientMode.Dynamic){
       initializeClientUsingConfigEndPoint(cf, addrs.get(0));
     } else {
@@ -352,14 +357,14 @@ private void initializeClientUsingConfigEndPoint(ConnectionFactory cf, InetSocke
     }
 
     //Initialize and start the poller.
-    configPoller = new ConfigurationPoller(this, cf.getDynamicModePollingInterval(), cf.isDaemon());
+    configPoller = new ConfigurationPoller(this, cf.getDynamicModePollingInterval(), cf.isDaemon(), configEndpointSelectionStrategy);
     configPoller.subscribeForClusterConfiguration(mconn);
   }
 
   private void setupConnection(ConnectionFactory cf, List<InetSocketAddress> addrs)
     throws IOException {
 
-    mconn = cf.createConnection(addrs);
+    mconn = configEndpointSelectionStrategy.setupMemcachedConnection(cf, addrs);
     assert mconn != null : "Connection factory failed to make a connection";
   }
 
@@ -1291,7 +1296,7 @@ public void complete() {
       }
     });
     rv.setOperation(op);
-    mconn.enqueueOperation(sa, op);
+    configEndpointSelectionStrategy.getConfigConnection().enqueueOperation(sa, op);
 
     return rv;
   }
@@ -1996,7 +2001,7 @@ public void complete() {
       }
     });
     rv.setOperation(op);
-    mconn.enqueueOperation(addr, op);
+    configEndpointSelectionStrategy.getConfigConnection().enqueueOperation(addr, op);
     return rv;
   }
 
@@ -2046,7 +2051,7 @@ public void complete() {
             }
           });
     rv.setOperation(op);
-    mconn.enqueueOperation(addr, op);
+    configEndpointSelectionStrategy.getConfigConnection().enqueueOperation(addr, op);
     return rv;
   }
 
@@ -2073,7 +2078,7 @@ public void complete() {
       }
     });
     rv.setOperation(op);
-    mconn.enqueueOperation(addr, op);
+    configEndpointSelectionStrategy.getConfigConnection().enqueueOperation(addr, op);
     return rv;
   }
 
@@ -2988,6 +2993,7 @@ public boolean shutdown(long timeout, TimeUnit unit) {
         mconn.setName(baseName + " - SHUTTING DOWN (telling client)");
         mconn.shutdown();
         mconn.setName(baseName + " - SHUTTING DOWN (informed client)");
+        configEndpointSelectionStrategy.shutdownConfigConnection();
         tcService.shutdown();
         //terminate all pending Auth Threads
         authMonitor.interruptAllPendingAuth();

src/main/java/net/spy/memcached/MemcachedClientIF.java

@@ -101,6 +101,16 @@ <T> CASResponse cas(String key, long casId, int exp, T value,
 
   Future<Boolean> replace(String key, int exp, Object o);
 
+  /**
+   * @return point-in-time view of currently available servers
+   */
+  Collection<net.spy.memcached.config.NodeEndPoint> getAvailableNodeEndPoints();
+
+  /**
+   * @return point-in-time view of all servers
+   */
+  Collection<net.spy.memcached.config.NodeEndPoint> getAllNodeEndPoints();
+
   <T> Future<T> asyncGet(String key, Transcoder<T> tc);
 
   Future<Object> asyncGet(String key);

src/main/java/net/spy/memcached/config/ConfigEndpointSelectionStrategy.java

@@ -0,0 +1,34 @@
+package net.spy.memcached.config;
+
+import net.spy.memcached.ConnectionFactory;
+import net.spy.memcached.MemcachedClientIF;
+import net.spy.memcached.MemcachedConnection;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.List;
+
+/**
+ * A configuration endpoint selection strategy which
+ * holds information and lifecycle relating to the
+ * configuration endpoint connection opened.
+ */
+public interface ConfigEndpointSelectionStrategy {
+    /**
+     * An endpoint is different from a connection;
+     * A connection could be composed of several endpoints
+     */
+    NodeEndPoint getConfigEndpoint(MemcachedClientIF client);
+    /**
+      * Delegate the responsibility of setting up a memcached connection to this class
+      * as, depending on the strategy, a copy of this memcached connection may be required
+      * to be created and kept alive
+      * @param cf connection factory used to build the Memcached connection
+      * @param addrs list of addresses that the connection factory should initialize connections with
+      * @return a Memcached connection to enqueue operations
+      * @throws IOException
+      */
+    MemcachedConnection setupMemcachedConnection(ConnectionFactory cf, List<InetSocketAddress> addrs) throws IOException;
+    void shutdownConfigConnection() throws IOException;
+    MemcachedConnection getConfigConnection();
+}

src/main/java/net/spy/memcached/config/RoundRobinConfigEndpointSelectionStrategy.java

@@ -0,0 +1,56 @@
+package net.spy.memcached.config;
+
+import net.spy.memcached.ConnectionFactory;
+import net.spy.memcached.MemcachedClientIF;
+import net.spy.memcached.MemcachedConnection;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+public class RoundRobinConfigEndpointSelectionStrategy implements ConfigEndpointSelectionStrategy {
+    private int currentIndex = 0;
+    private MemcachedConnection configurationConnection;
+
+    @Override
+    public NodeEndPoint getConfigEndpoint(MemcachedClientIF client) {
+        Collection<NodeEndPoint> endpoints = client.getAvailableNodeEndPoints();
+        if (endpoints.isEmpty()) {
+            //If no nodes are available status, then get all the endpoints. This provides an
+            //opportunity to re-resolve the hostname by recreating InetSocketAddress instance in "NodeEndPoint".getInetSocketAddress().
+            endpoints = client.getAllNodeEndPoints();
+        }
+        currentIndex = (currentIndex + 1) % endpoints.size();
+        Iterator<NodeEndPoint> iterator = endpoints.iterator();
+        for (int i = 0; i < currentIndex; i++) {
+            iterator.next();
+        }
+        return iterator.next();
+    }
+
+    /**
+     * The Round Robin strategy (AWS default) reuses the same connection to retrieve configuration.
+     * In this case, we only need to hold a reference to the connection to return when
+     * `getConfigConnection` is called
+     *
+     */
+    @Override
+    public MemcachedConnection setupMemcachedConnection(ConnectionFactory cf, List<InetSocketAddress> addrs) throws IOException {
+        configurationConnection = cf.createConnection(addrs);
+        return configurationConnection;
+    }
+
+    @Override
+    public void shutdownConfigConnection() {
+        // As the Memcached connection doubles as a configuration connection,
+        // this operation is a no-op as that should be handled by the consumer of the
+        // Memcached connection
+    }
+
+    @Override
+    public MemcachedConnection getConfigConnection() {
+        return configurationConnection;
+    }
+}