[automatic failover] Introduce fast failover mode - a thread-sync-free approach

.gitignore

@@ -17,3 +17,4 @@ tags
 redis-git
 appendonlydir/
 .DS_Store
+.vscode/settings.json

pom.xml

@@ -143,7 +143,7 @@
 		<dependency>
 			<groupId>ch.qos.logback</groupId>
 			<artifactId>logback-classic</artifactId>
-			<version>1.3.15</version>
+			<version>1.2.12</version>
 			<scope>test</scope>
 		</dependency>
 		<dependency>

src/main/java/redis/clients/jedis/CommandObjects.java

@@ -74,6 +74,10 @@ public final CommandObject<String> ping() {
     return PING_COMMAND_OBJECT;
   }
 
+  public final CommandObject<String> echo(String msg) {
+    return new CommandObject<>(commandArguments(ECHO).add(msg), BuilderFactory.STRING);
+  }
+
   private final CommandObject<String> FLUSHALL_COMMAND_OBJECT = new CommandObject<>(commandArguments(FLUSHALL), BuilderFactory.STRING);
 
   public final CommandObject<String> flushAll() {

src/main/java/redis/clients/jedis/Connection.java

@@ -33,6 +33,47 @@
 
 public class Connection implements Closeable {
 
+  public static class Builder {
+    private JedisSocketFactory socketFactory;
+    private JedisClientConfig clientConfig;
+    private InitializationTracker<Connection> tracker;
+
+    public Builder setSocketFactory(JedisSocketFactory socketFactory) {
+      this.socketFactory = socketFactory;
+      return this;
+    }
+
+    public Builder setClientConfig(JedisClientConfig clientConfig) {
+      this.clientConfig = clientConfig;
+      return this;
+    }
+
+    public Builder setTracker(InitializationTracker<Connection> tracker) {
+      this.tracker = tracker;
+      return this;
+    }
+
+    public JedisSocketFactory getSocketFactory() {
+      return socketFactory;
+    }
+
+    public JedisClientConfig getClientConfig() {
+      return clientConfig;
+    }
+
+    public InitializationTracker<Connection> getTracker() {
+      return tracker;
+    }
+
+    public Connection build() {
+      return new Connection(this);
+    }
+  }
+
+  public static Builder builder(){
+    return new Builder();
+  }
+
   private ConnectionPool memberOf;
   protected RedisProtocol protocol;
   private final JedisSocketFactory socketFactory;
@@ -72,11 +113,25 @@ public Connection(final JedisSocketFactory socketFactory) {
 
   public Connection(final JedisSocketFactory socketFactory, JedisClientConfig clientConfig) {
     this.socketFactory = socketFactory;
-    this.soTimeout = clientConfig.getSocketTimeoutMillis();
-    this.infiniteSoTimeout = clientConfig.getBlockingSocketTimeoutMillis();
     initializeFromClientConfig(clientConfig);
   }
 
+  protected Connection(Builder builder) {
+    this.socketFactory = builder.getSocketFactory();
+    InitializationTracker<Connection> tracker = builder.getTracker();
+
+    if (tracker != null) {
+      tracker.add(this);
+      try {
+        initializeFromClientConfig(builder.getClientConfig());
+      } finally {
+        tracker.remove(this);
+      }
+    } else {
+      initializeFromClientConfig(builder.getClientConfig());
+    }
+  }
+
   @Override
   public String toString() {
     return getClass().getSimpleName() + "{" + socketFactory + "}";
@@ -288,6 +343,10 @@ public void disconnect() {
     }
   }
 
+  public void forceDisconnect() throws IOException {
+    socket.close();
+  }
+
   public boolean isConnected() {
     return socket != null && socket.isBound() && !socket.isClosed() && socket.isConnected()
         && !socket.isInputShutdown() && !socket.isOutputShutdown();
@@ -452,6 +511,9 @@ private static boolean validateClientInfo(String info) {
 
   protected void initializeFromClientConfig(final JedisClientConfig config) {
     try {
+      this.soTimeout = config.getSocketTimeoutMillis();
+      this.infiniteSoTimeout = config.getBlockingSocketTimeoutMillis();
+
       connect();
 
       protocol = config.getRedisProtocol();

src/main/java/redis/clients/jedis/ConnectionFactory.java

@@ -7,6 +7,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.function.Supplier;
+import java.util.function.UnaryOperator;
 
 import redis.clients.jedis.annots.Experimental;
 import redis.clients.jedis.authentication.AuthXManager;
@@ -21,14 +22,75 @@
  */
 public class ConnectionFactory implements PooledObjectFactory<Connection> {
 
+  public static class Builder {
+
+    private JedisSocketFactory jedisSocketFactory;
+    private JedisClientConfig clientConfig;
+    private Cache cache;
+    private InitializationTracker<Connection> tracker;
+    private HostAndPort hostAndPort;
+
+    public JedisSocketFactory getJedisSocketFactory() {
+      return jedisSocketFactory;
+    }
+
+    public JedisClientConfig getClientConfig() {
+      return clientConfig;
+    }
+
+    public Cache getCache() {
+      return cache;
+    }
+
+    public InitializationTracker<Connection> getTracker() {
+      return tracker;
+    }
+
+    public Builder setJedisSocketFactory(JedisSocketFactory jedisSocketFactory) {
+      this.jedisSocketFactory = jedisSocketFactory;
+      return this;
+    }
+
+    public Builder setClientConfig(JedisClientConfig clientConfig) {
+      this.clientConfig = clientConfig;
+      return this;
+    }
+
+    public Builder setCache(Cache cache) {
+      this.cache = cache;
+      return this;
+    }
+
+    public Builder setTracker(InitializationTracker<Connection> tracker) {
+      this.tracker = tracker;
+      return this;
+    }
+
+    public Builder setHostAndPort(HostAndPort hostAndPort) {
+      this.hostAndPort = hostAndPort;
+      return this;
+    }
+
+    public ConnectionFactory build() {
+      return new ConnectionFactory(this);
+    }
+
+  }
+
+  public static Builder builder() {
+    return new Builder();
+  }
+
   private static final Logger logger = LoggerFactory.getLogger(ConnectionFactory.class);
 
   private final JedisSocketFactory jedisSocketFactory;
   private final JedisClientConfig clientConfig;
   private final Cache clientSideCache;
-  private final Supplier<Connection> objectMaker;
+  private Supplier<Connection> objectMaker;
+
+  private AuthXEventListener authXEventListener;
 
-  private final AuthXEventListener authXEventListener;
+  private InitializationTracker<Connection> tracker;
 
   public ConnectionFactory(final HostAndPort hostAndPort) {
     this(hostAndPort, DefaultJedisClientConfig.builder().build(), null);
@@ -56,6 +118,24 @@ private ConnectionFactory(final JedisSocketFactory jedisSocketFactory,
     this.clientSideCache = csCache;
     this.clientConfig = clientConfig;
 
+    initAuthXManager();
+  }
+
+  public ConnectionFactory(Builder builder) {
+    this.clientConfig = builder.getClientConfig() != null ? builder.getClientConfig()
+      : DefaultJedisClientConfig.builder().build();
+    if (builder.getJedisSocketFactory() == null) {
+      this.jedisSocketFactory = new DefaultJedisSocketFactory(builder.hostAndPort, this.clientConfig);
+    } else {
+      this.jedisSocketFactory = builder.getJedisSocketFactory();
+    }
+    this.clientSideCache = builder.getCache();
+    this.tracker = builder.getTracker();
+
+    initAuthXManager();
+  }
+
+  private void initAuthXManager() {
     AuthXManager authXManager = clientConfig.getAuthXManager();
     if (authXManager == null) {
       this.objectMaker = connectionSupplier();
@@ -69,8 +149,13 @@ private ConnectionFactory(final JedisSocketFactory jedisSocketFactory,
   }
 
   private Supplier<Connection> connectionSupplier() {
-    return clientSideCache == null ? () -> new Connection(jedisSocketFactory, clientConfig)
-        : () -> new CacheConnection(jedisSocketFactory, clientConfig, clientSideCache);
+    Connection.Builder conBuilder = clientSideCache == null ? Connection.builder()
+      : CacheConnection.builder(clientSideCache);
+    conBuilder.setSocketFactory(jedisSocketFactory).setClientConfig(clientConfig);
+    if (tracker != null) {
+      conBuilder.setTracker(tracker);
+    }
+    return () -> conBuilder.build();
   }
 
   @Override

src/main/java/redis/clients/jedis/ConnectionPool.java

@@ -3,6 +3,7 @@
 import org.apache.commons.pool2.PooledObjectFactory;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 
+import redis.clients.authentication.core.Token;
 import redis.clients.jedis.annots.Experimental;
 import redis.clients.jedis.authentication.AuthXManager;
 import redis.clients.jedis.csc.Cache;
@@ -65,17 +66,25 @@ public void close() {
     }
   }
 
-  private void attachAuthenticationListener(AuthXManager authXManager) {
+  protected void attachAuthenticationListener(AuthXManager authXManager) {
     this.authXManager = authXManager;
     if (authXManager != null) {
-      authXManager.addPostAuthenticationHook(token -> {
-        try {
-          // this is to trigger validations on each connection via ConnectionFactory
-          evict();
-        } catch (Exception e) {
-          throw new JedisException("Failed to evict connections from pool", e);
-        }
-      });
+      authXManager.addPostAuthenticationHook(this::postAuthentication);
+    }
+  }
+
+  protected void detachAuthenticationListener() {
+    if (authXManager != null) {
+      authXManager.removePostAuthenticationHook(this::postAuthentication);
+    }
+  }
+
+  private void postAuthentication(Token token) {
+    try {
+      // this is to trigger validations on each connection via ConnectionFactory
+      evict();
+    } catch (Exception e) {
+      throw new JedisException("Failed to evict connections from pool", e);
     }
   }
 }

src/main/java/redis/clients/jedis/HostAndPort.java

@@ -2,7 +2,9 @@
 
 import java.io.Serializable;
 
-public class HostAndPort implements Serializable {
+import redis.clients.jedis.mcf.Endpoint;
+
+public class HostAndPort implements Serializable, Endpoint {
 
     private static final long serialVersionUID = -519876229978427751L;
 
@@ -14,10 +16,12 @@ public HostAndPort(String host, int port) {
         this.port = port;
     }
 
+    @Override
     public String getHost() {
         return host;
     }
 
+    @Override
     public int getPort() {
         return port;
     }

src/main/java/redis/clients/jedis/InitializationTracker.java

@@ -0,0 +1,21 @@
+package redis.clients.jedis;
+
+public interface InitializationTracker<T> extends Iterable<T> {
+    void add(T target);
+    void remove(T target);
+
+    public static InitializationTracker NOOP = new InitializationTracker() {
+        @Override
+        public void add(Object target) {
+        }
+
+        @Override
+        public void remove(Object target) {
+        }
+
+        @Override
+        public java.util.Iterator iterator() {
+            return java.util.Collections.emptyIterator();
+        }
+    };
+}