Add support to the use of JedisSocketFactory using a pool (#2293)

* Add support to the use of JedisSocketFactory using a pool

- Support for JedisSocketFactory has already been added to the lowest level Jedis to support adding any custom socket factory (e.g. UDS), this propagates the support in the JedisPool too
- Also fix Jedis/BinaryJedis constructors that broke after the introduction of config due to missing client initialization

* Rework host and port updates through the socket factory

* Add a new constructor to DefaultJedisSocketFactory that accepts a JedisClientConfig and cleanup JedisFactory

* Only set host and port if we pass a non-null value

* Update Deprecation JavaDoc

* Use a volatile instead of atomic reference since we are not doing any CAS operations

Co-authored-by: Mina Asham <minaasha@amazon.com>
Co-authored-by: M Sazzadul Hoque <7600764+sazzad16@users.noreply.github.com>

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

@@ -278,10 +278,21 @@ private void initializeFromURI(URI uri) {
     }
   }
 
+  /**
+   * @deprecated This constructor will be removed in future major release.
+   *
+   * Use {@link BinaryJedis#BinaryJedis(redis.clients.jedis.JedisSocketFactory, redis.clients.jedis.JedisClientConfig)}.
+   */
+  @Deprecated
   public BinaryJedis(final JedisSocketFactory jedisSocketFactory) {
     client = new Client(jedisSocketFactory);
   }
 
+  public BinaryJedis(final JedisSocketFactory jedisSocketFactory, final JedisClientConfig clientConfig) {
+    client = new Client(jedisSocketFactory);
+    initializeFromClientConfig(clientConfig);
+  }
+
   public boolean isConnected() {
     return client.isConnected();
   }

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

@@ -16,7 +16,7 @@ public class DefaultJedisSocketFactory implements JedisSocketFactory {
   protected static final HostAndPort DEFAULT_HOST_AND_PORT = new HostAndPort(Protocol.DEFAULT_HOST,
       Protocol.DEFAULT_PORT);
 
-  private HostAndPort hostAndPort = DEFAULT_HOST_AND_PORT;
+  private volatile HostAndPort hostAndPort = DEFAULT_HOST_AND_PORT;
   private int connectionTimeout = Protocol.DEFAULT_TIMEOUT;
   private int socketTimeout = Protocol.DEFAULT_TIMEOUT;
   private boolean ssl = false;
@@ -32,6 +32,10 @@ public DefaultJedisSocketFactory(HostAndPort hostAndPort) {
     this(hostAndPort, null);
   }
 
+  public DefaultJedisSocketFactory(JedisClientConfig config) {
+    this(null, config);
+  }
+
   @Deprecated
   public DefaultJedisSocketFactory(String host, int port, int connectionTimeout, int socketTimeout,
       boolean ssl, SSLSocketFactory sslSocketFactory, SSLParameters sslParameters,
@@ -46,7 +50,9 @@ public DefaultJedisSocketFactory(String host, int port, int connectionTimeout, i
   }
 
   public DefaultJedisSocketFactory(HostAndPort hostAndPort, JedisClientConfig config) {
-    this.hostAndPort = hostAndPort;
+    if (hostAndPort != null) {
+      this.hostAndPort = hostAndPort;
+    }
     if (config != null) {
       this.connectionTimeout = config.getConnectionTimeoutMillis();
       this.socketTimeout = config.getSocketTimeoutMillis();
@@ -105,6 +111,11 @@ public Socket createSocket() throws JedisConnectionException {
     }
   }
 
+  @Override
+  public void updateHostAndPort(HostAndPort hostAndPort) {
+    this.hostAndPort = hostAndPort;
+  }
+
   public HostAndPort getSocketHostAndPort() {
     HostAndPortMapper mapper = getHostAndPortMapper();
     HostAndPort hostAndPort = getHostAndPort();

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

@@ -153,10 +153,20 @@ public Jedis(final URI uri, JedisClientConfig config) {
     super(uri, config);
   }
 
+  /**
+   * @deprecated This constructor will be removed in future major release.
+   *
+   * Use {@link Jedis#Jedis(redis.clients.jedis.JedisSocketFactory, redis.clients.jedis.JedisClientConfig)}.
+   */
+  @Deprecated
   public Jedis(final JedisSocketFactory jedisSocketFactory) {
     super(jedisSocketFactory);
   }
 
+  public Jedis(final JedisSocketFactory jedisSocketFactory, final JedisClientConfig clientConfig) {
+    super(jedisSocketFactory, clientConfig);
+  }
+
   /**
    * COPY source destination [DB destination-db] [REPLACE]
    *

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

@@ -1,7 +1,6 @@
 package redis.clients.jedis;
 
 import java.net.URI;
-import java.util.concurrent.atomic.AtomicReference;
 
 import javax.net.ssl.HostnameVerifier;
 import javax.net.ssl.SSLParameters;
@@ -24,7 +23,7 @@ public class JedisFactory implements PooledObjectFactory<Jedis> {
 
   private static final Logger logger = LoggerFactory.getLogger(JedisFactory.class);
 
-  private final AtomicReference<HostAndPort> hostAndPort = new AtomicReference<>();
+  private final JedisSocketFactory jedisSocketFactory;
 
   private final JedisClientConfig clientConfig;
 
@@ -66,20 +65,25 @@ protected JedisFactory(final String host, final int port, final int connectionTi
   }
 
   protected JedisFactory(final HostAndPort hostAndPort, final JedisClientConfig clientConfig) {
-    this.hostAndPort.set(hostAndPort);
     this.clientConfig = DefaultJedisClientConfig.copyConfig(clientConfig);
+    this.jedisSocketFactory = new DefaultJedisSocketFactory(hostAndPort, this.clientConfig);
   }
 
   protected JedisFactory(final String host, final int port, final int connectionTimeout, final int soTimeout,
       final int infiniteSoTimeout, final String user, final String password, final int database,
       final String clientName, final boolean ssl, final SSLSocketFactory sslSocketFactory,
       final SSLParameters sslParameters, final HostnameVerifier hostnameVerifier) {
-    this.hostAndPort.set(new HostAndPort(host, port));
     this.clientConfig = DefaultJedisClientConfig.builder().connectionTimeoutMillis(connectionTimeout)
         .socketTimeoutMillis(soTimeout).blockingSocketTimeoutMillis(infiniteSoTimeout).user(user)
         .password(password).database(database).clientName(clientName)
         .ssl(ssl).sslSocketFactory(sslSocketFactory)
         .sslParameters(sslParameters).hostnameVerifier(hostnameVerifier).build();
+    this.jedisSocketFactory = new DefaultJedisSocketFactory(new HostAndPort(host, port), this.clientConfig);
+  }
+
+  protected JedisFactory(final JedisSocketFactory jedisSocketFactory, final JedisClientConfig clientConfig) {
+    this.clientConfig = DefaultJedisClientConfig.copyConfig(clientConfig);
+    this.jedisSocketFactory = jedisSocketFactory;
   }
 
   /**
@@ -100,6 +104,7 @@ protected JedisFactory(final int connectionTimeout, final int soTimeout, final i
    */
   protected JedisFactory(final JedisClientConfig clientConfig) {
     this.clientConfig = clientConfig;
+    this.jedisSocketFactory = new DefaultJedisSocketFactory(clientConfig);
   }
 
   protected JedisFactory(final URI uri, final int connectionTimeout, final int soTimeout,
@@ -120,17 +125,17 @@ protected JedisFactory(final URI uri, final int connectionTimeout, final int soT
       throw new InvalidURIException(String.format(
           "Cannot open Redis connection due invalid URI. %s", uri.toString()));
     }
-    this.hostAndPort.set(new HostAndPort(uri.getHost(), uri.getPort()));
     this.clientConfig = DefaultJedisClientConfig.builder().connectionTimeoutMillis(connectionTimeout)
         .socketTimeoutMillis(soTimeout).blockingSocketTimeoutMillis(infiniteSoTimeout)
         .user(JedisURIHelper.getUser(uri)).password(JedisURIHelper.getPassword(uri))
         .database(JedisURIHelper.getDBIndex(uri)).clientName(clientName)
         .ssl(JedisURIHelper.isRedisSSLScheme(uri)).sslSocketFactory(sslSocketFactory)
         .sslParameters(sslParameters).hostnameVerifier(hostnameVerifier).build();
+    this.jedisSocketFactory = new DefaultJedisSocketFactory(new HostAndPort(uri.getHost(), uri.getPort()), this.clientConfig);
   }
 
   public void setHostAndPort(final HostAndPort hostAndPort) {
-    this.hostAndPort.set(hostAndPort);
+    jedisSocketFactory.updateHostAndPort(hostAndPort);
   }
 
   public void setPassword(final String password) {
@@ -167,10 +172,9 @@ public void destroyObject(PooledObject<Jedis> pooledJedis) throws Exception {
 
   @Override
   public PooledObject<Jedis> makeObject() throws Exception {
-    final HostAndPort hostPort = this.hostAndPort.get();
     Jedis jedis = null;
     try {
-      jedis = new Jedis(hostPort, clientConfig);
+      jedis = new Jedis(jedisSocketFactory, clientConfig);
       jedis.connect();
       return new DefaultPooledObject<>(jedis);
     } catch (JedisException je) {
@@ -199,13 +203,14 @@ public void passivateObject(PooledObject<Jedis> pooledJedis) throws Exception {
   public boolean validateObject(PooledObject<Jedis> pooledJedis) {
     final BinaryJedis jedis = pooledJedis.getObject();
     try {
-      HostAndPort hostAndPort = this.hostAndPort.get();
+      String host = jedisSocketFactory.getHost();
+      int port = jedisSocketFactory.getPort();
 
       String connectionHost = jedis.getClient().getHost();
       int connectionPort = jedis.getClient().getPort();
 
-      return hostAndPort.getHost().equals(connectionHost)
-          && hostAndPort.getPort() == connectionPort && jedis.isConnected()
+      return host.equals(connectionHost)
+          && port == connectionPort && jedis.isConnected()
           && jedis.ping().equals("PONG");
     } catch (final Exception e) {
       return false;

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

@@ -261,6 +261,11 @@ public JedisPool(final GenericObjectPoolConfig<Jedis> poolConfig, final HostAndP
     super(poolConfig, new JedisFactory(hostAndPort, clientConfig));
   }
 
+  public JedisPool(final GenericObjectPoolConfig<Jedis> poolConfig, final JedisSocketFactory jedisSocketFactory,
+      final JedisClientConfig clientConfig) {
+    super(poolConfig, new JedisFactory(jedisSocketFactory, clientConfig));
+  }
+
   public JedisPool(final GenericObjectPoolConfig<Jedis> poolConfig) {
     this(poolConfig, Protocol.DEFAULT_HOST, Protocol.DEFAULT_PORT);
   }

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

@@ -24,6 +24,8 @@ public interface JedisSocketFactory {
    */
   Socket createSocket() throws IOException, JedisConnectionException;
 
+  void updateHostAndPort(HostAndPort hostAndPort);
+
   @Deprecated
   String getDescription();
 

src/test/java/redis/clients/jedis/tests/UdsTest.java

@@ -7,6 +7,7 @@
 import org.newsclub.net.unix.AFUNIXSocket;
 import org.newsclub.net.unix.AFUNIXSocketAddress;
 
+import redis.clients.jedis.HostAndPort;
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.JedisSocketFactory;
 import redis.clients.jedis.Protocol;
@@ -38,6 +39,11 @@ public Socket createSocket() throws JedisConnectionException {
       }
     }
 
+    @Override
+    public void updateHostAndPort(HostAndPort hostAndPort) {
+      throw new UnsupportedOperationException("UDS cannot update host and port");
+    }
+
     @Override
     public String getDescription() {
       return UDS_SOCKET.toString();