Easier constructors for pipeline classes (#2731)

redis · Dec 12, 2021 · 022a9af · 022a9af
1 parent c0136df
commit 022a9af
Show file tree

Hide file tree

Showing 8 changed files with 691 additions and 498 deletions.
diff --git a/src/main/java/redis/clients/jedis/ClusterPipeline.java b/src/main/java/redis/clients/jedis/ClusterPipeline.java
@@ -1,16 +1,40 @@
 package redis.clients.jedis;
 
+import java.util.Set;
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import redis.clients.jedis.providers.ClusterConnectionProvider;
+import redis.clients.jedis.util.IOUtils;
 
 public class ClusterPipeline extends MultiNodePipelineBase {
 
   private final ClusterConnectionProvider provider;
+  private AutoCloseable closeable = null;
+
+  public ClusterPipeline(Set<HostAndPort> clusterNodes, JedisClientConfig clientConfig) {
+    this(new ClusterConnectionProvider(clusterNodes, clientConfig));
+    this.closeable = this.provider;
+  }
+
+  public ClusterPipeline(Set<HostAndPort> clusterNodes, JedisClientConfig clientConfig,
+      GenericObjectPoolConfig<Connection> poolConfig) {
+    this(new ClusterConnectionProvider(clusterNodes, clientConfig, poolConfig));
+    this.closeable = this.provider;
+  }
 
   public ClusterPipeline(ClusterConnectionProvider provider) {
     super(new ClusterCommandObjects());
     this.provider = provider;
   }
 
+  @Override
+  public void close() {
+    try {
+      super.close();
+    } finally {
+      IOUtils.closeQuietly(closeable);
+    }
+  }
+
   @Override
   protected HostAndPort getNodeKey(CommandArguments args) {
     return provider.getNode(((ClusterCommandArguments) args).getCommandHashSlot());

diff --git a/src/main/java/redis/clients/jedis/JedisCluster.java b/src/main/java/redis/clients/jedis/JedisCluster.java
@@ -108,80 +108,80 @@ public JedisCluster(Set<HostAndPort> nodes, int timeout,
     this(nodes, timeout, DEFAULT_MAX_ATTEMPTS, poolConfig);
   }
 
-  public JedisCluster(Set<HostAndPort> jedisClusterNode, int timeout, int maxAttempts,
+  public JedisCluster(Set<HostAndPort> clusterNodes, int timeout, int maxAttempts,
       final GenericObjectPoolConfig<Connection> poolConfig) {
-    this(jedisClusterNode, timeout, timeout, maxAttempts, poolConfig);
+    this(clusterNodes, timeout, timeout, maxAttempts, poolConfig);
   }
 
-  public JedisCluster(Set<HostAndPort> jedisClusterNode, int connectionTimeout,
-      int soTimeout, int maxAttempts, final GenericObjectPoolConfig<Connection> poolConfig) {
-    this(jedisClusterNode, connectionTimeout, soTimeout, maxAttempts, null, poolConfig);
+  public JedisCluster(Set<HostAndPort> clusterNodes, int connectionTimeout, int soTimeout,
+      int maxAttempts, final GenericObjectPoolConfig<Connection> poolConfig) {
+    this(clusterNodes, connectionTimeout, soTimeout, maxAttempts, null, poolConfig);
   }
 
-  public JedisCluster(Set<HostAndPort> jedisClusterNode, int connectionTimeout,
-      int soTimeout, int maxAttempts, String password, GenericObjectPoolConfig<Connection> poolConfig) {
-    this(jedisClusterNode, connectionTimeout, soTimeout, maxAttempts, password, null, poolConfig);
+  public JedisCluster(Set<HostAndPort> clusterNodes, int connectionTimeout, int soTimeout,
+      int maxAttempts, String password, GenericObjectPoolConfig<Connection> poolConfig) {
+    this(clusterNodes, connectionTimeout, soTimeout, maxAttempts, password, null, poolConfig);
   }
 
-  public JedisCluster(Set<HostAndPort> jedisClusterNode, int connectionTimeout,
+  public JedisCluster(Set<HostAndPort> clusterNodes, int connectionTimeout,
       int soTimeout, int maxAttempts, String password, String clientName,
       GenericObjectPoolConfig<Connection> poolConfig) {
-    this(jedisClusterNode, connectionTimeout, soTimeout, maxAttempts, null, password, clientName,
+    this(clusterNodes, connectionTimeout, soTimeout, maxAttempts, null, password, clientName,
         poolConfig);
   }
 
-  public JedisCluster(Set<HostAndPort> jedisClusterNode, int connectionTimeout,
-      int soTimeout, int maxAttempts, String user, String password, String clientName,
+  public JedisCluster(Set<HostAndPort> clusterNodes, int connectionTimeout, int soTimeout,
+      int maxAttempts, String user, String password, String clientName,
       GenericObjectPoolConfig<Connection> poolConfig) {
-    this(jedisClusterNode, DefaultJedisClientConfig.builder().connectionTimeoutMillis(connectionTimeout)
+    this(clusterNodes, DefaultJedisClientConfig.builder().connectionTimeoutMillis(connectionTimeout)
         .socketTimeoutMillis(soTimeout).user(user).password(password).clientName(clientName).build(),
         maxAttempts, poolConfig);
   }
 
-  public JedisCluster(Set<HostAndPort> jedisClusterNode, int connectionTimeout,
+  public JedisCluster(Set<HostAndPort> clusterNodes, int connectionTimeout,
       int soTimeout, int infiniteSoTimeout, int maxAttempts, String user, String password,
       String clientName, GenericObjectPoolConfig<Connection> poolConfig) {
-    this(jedisClusterNode, DefaultJedisClientConfig.builder().connectionTimeoutMillis(connectionTimeout)
+    this(clusterNodes, DefaultJedisClientConfig.builder().connectionTimeoutMillis(connectionTimeout)
         .socketTimeoutMillis(soTimeout).blockingSocketTimeoutMillis(infiniteSoTimeout)
         .user(user).password(password).clientName(clientName).build(), maxAttempts, poolConfig);
   }
 
-  public JedisCluster(Set<HostAndPort> jedisClusterNode, int connectionTimeout,
-      int soTimeout, int maxAttempts, String password, String clientName,
+  public JedisCluster(Set<HostAndPort> clusterNodes, int connectionTimeout, int soTimeout,
+      int maxAttempts, String password, String clientName,
       GenericObjectPoolConfig<Connection> poolConfig, boolean ssl) {
-    this(jedisClusterNode, connectionTimeout, soTimeout, maxAttempts, null, password, clientName,
+    this(clusterNodes, connectionTimeout, soTimeout, maxAttempts, null, password, clientName,
         poolConfig, ssl);
   }
 
-  public JedisCluster(Set<HostAndPort> jedisClusterNode, int connectionTimeout,
-      int soTimeout, int maxAttempts, String user, String password, String clientName,
+  public JedisCluster(Set<HostAndPort> clusterNodes, int connectionTimeout, int soTimeout,
+      int maxAttempts, String user, String password, String clientName,
       GenericObjectPoolConfig<Connection> poolConfig, boolean ssl) {
-    this(jedisClusterNode, DefaultJedisClientConfig.builder().connectionTimeoutMillis(connectionTimeout)
+    this(clusterNodes, DefaultJedisClientConfig.builder().connectionTimeoutMillis(connectionTimeout)
         .socketTimeoutMillis(soTimeout).user(user).password(password).clientName(clientName).ssl(ssl).build(),
         maxAttempts, poolConfig);
   }
 
-  public JedisCluster(Set<HostAndPort> jedisClusterNode, JedisClientConfig clientConfig,
+  public JedisCluster(Set<HostAndPort> clusterNodes, JedisClientConfig clientConfig,
       int maxAttempts, GenericObjectPoolConfig<Connection> poolConfig) {
-    this(jedisClusterNode, clientConfig, maxAttempts,
+    this(clusterNodes, clientConfig, maxAttempts,
         Duration.ofMillis((long) clientConfig.getSocketTimeoutMillis() * maxAttempts), poolConfig);
   }
 
-  public JedisCluster(Set<HostAndPort> jedisClusterNode, JedisClientConfig clientConfig,
+  public JedisCluster(Set<HostAndPort> clusterNodes, JedisClientConfig clientConfig,
       int maxAttempts, Duration maxTotalRetriesDuration, GenericObjectPoolConfig<Connection> poolConfig) {
-    super(jedisClusterNode, clientConfig, poolConfig, maxAttempts, maxTotalRetriesDuration);
+    super(clusterNodes, clientConfig, poolConfig, maxAttempts, maxTotalRetriesDuration);
   }
 
-  public JedisCluster(Set<HostAndPort> jedisClusterNodes, JedisClientConfig clientConfig) {
-    this(jedisClusterNodes, clientConfig, DEFAULT_MAX_ATTEMPTS);
+  public JedisCluster(Set<HostAndPort> clusterNodess, JedisClientConfig clientConfig) {
+    this(clusterNodess, clientConfig, DEFAULT_MAX_ATTEMPTS);
   }
 
-  public JedisCluster(Set<HostAndPort> jedisClusterNodes, JedisClientConfig clientConfig, int maxAttempts) {
-    super(jedisClusterNodes, clientConfig, maxAttempts);
+  public JedisCluster(Set<HostAndPort> clusterNodess, JedisClientConfig clientConfig, int maxAttempts) {
+    super(clusterNodess, clientConfig, maxAttempts);
   }
 
-  public JedisCluster(Set<HostAndPort> jedisClusterNodes, JedisClientConfig clientConfig, int maxAttempts, Duration maxTotalRetriesDuration) {
-    super(jedisClusterNodes, clientConfig, maxAttempts, maxTotalRetriesDuration);
+  public JedisCluster(Set<HostAndPort> clusterNodess, JedisClientConfig clientConfig, int maxAttempts, Duration maxTotalRetriesDuration) {
+    super(clusterNodess, clientConfig, maxAttempts, maxTotalRetriesDuration);
   }
 
   public Map<String, ConnectionPool> getClusterNodes() {

diff --git a/src/main/java/redis/clients/jedis/MultiNodePipelineBase.java b/src/main/java/redis/clients/jedis/MultiNodePipelineBase.java
@@ -63,7 +63,7 @@ protected final <T> Response<T> appendCommand(CommandObject<T> commandObject) {
   }
 
   @Override
-  public final void close() {
+  public void close() {
     sync();
     for (Connection connection : connections.values()) {
       connection.close();

diff --git a/src/main/java/redis/clients/jedis/ShardedPipeline.java b/src/main/java/redis/clients/jedis/ShardedPipeline.java
@@ -1,22 +1,47 @@
 package redis.clients.jedis;
 
-import redis.clients.jedis.providers.ShardedConnectionProvider;
+import java.util.List;
 import java.util.regex.Pattern;
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import redis.clients.jedis.providers.ShardedConnectionProvider;
+import redis.clients.jedis.util.Hashing;
+import redis.clients.jedis.util.IOUtils;
 
 public class ShardedPipeline extends MultiNodePipelineBase {
 
   private final ShardedConnectionProvider provider;
+  private AutoCloseable closeable = null;
+
+  public ShardedPipeline(List<HostAndPort> shards, JedisClientConfig clientConfig) {
+    this(new ShardedConnectionProvider(shards, clientConfig));
+    this.closeable = this.provider;
+  }
 
   public ShardedPipeline(ShardedConnectionProvider provider) {
     super(new ShardedCommandObjects(provider.getHashingAlgo()));
     this.provider = provider;
   }
 
+  public ShardedPipeline(List<HostAndPort> shards, JedisClientConfig clientConfig,
+      GenericObjectPoolConfig<Connection> poolConfig, Hashing algo, Pattern tagPattern) {
+    this(new ShardedConnectionProvider(shards, clientConfig, poolConfig, algo), tagPattern);
+    this.closeable = this.provider;
+  }
+
   public ShardedPipeline(ShardedConnectionProvider provider, Pattern tagPattern) {
     super(new ShardedCommandObjects(provider.getHashingAlgo(), tagPattern));
     this.provider = provider;
   }
 
+  @Override
+  public void close() {
+    try {
+      super.close();
+    } finally {
+      IOUtils.closeQuietly(closeable);
+    }
+  }
+
   @Override
   protected HostAndPort getNodeKey(CommandArguments args) {
     return provider.getNode(((ShardedCommandArguments) args).getKeyHash());

diff --git a/src/main/java/redis/clients/jedis/providers/ClusterConnectionProvider.java b/src/main/java/redis/clients/jedis/providers/ClusterConnectionProvider.java
@@ -22,15 +22,15 @@ public class ClusterConnectionProvider implements ConnectionProvider {
 
   protected final JedisClusterInfoCache cache;
 
-  public ClusterConnectionProvider(Set<HostAndPort> jedisClusterNodes, JedisClientConfig clientConfig) {
+  public ClusterConnectionProvider(Set<HostAndPort> clusterNodes, JedisClientConfig clientConfig) {
     this.cache = new JedisClusterInfoCache(clientConfig);
-    initializeSlotsCache(jedisClusterNodes, clientConfig);
+    initializeSlotsCache(clusterNodes, clientConfig);
   }
 
-  public ClusterConnectionProvider(Set<HostAndPort> jedisClusterNodes, JedisClientConfig clientConfig,
+  public ClusterConnectionProvider(Set<HostAndPort> clusterNodes, JedisClientConfig clientConfig,
       GenericObjectPoolConfig<Connection> poolConfig) {
     this.cache = new JedisClusterInfoCache(clientConfig, poolConfig);
-    initializeSlotsCache(jedisClusterNodes, clientConfig);
+    initializeSlotsCache(clusterNodes, clientConfig);
   }
 
   private void initializeSlotsCache(Set<HostAndPort> startNodes, JedisClientConfig clientConfig) {