Reuse stale connections collection when closing stale connections #443

Stale connections expiry now uses the same collection of connections to prevent stale expiry preparation and the actual closing to use different connections.

src/main/java/com/lambdaworks/redis/cluster/PooledClusterConnectionProvider.java

@@ -306,17 +306,22 @@ private void reconfigurePartitions() {
         }
 
         resetFastConnectionCache();
-        closeStaleConnections();
+        closeStaleConnections(staleConnections);
     }
 
     /**
      * Close stale connections.
      */
     @Override
     public void closeStaleConnections() {
-        logger.debug("closeStaleConnections() count before expiring: {}", getConnectionCount());
 
         Set<ConnectionKey> stale = getStaleConnectionKeys();
+        closeStaleConnections(stale);
+    }
+
+    protected void closeStaleConnections(Set<ConnectionKey> stale) {
+
+        logger.debug("closeStaleConnections() count before expiring: {}", getConnectionCount());
 
         for (ConnectionKey connectionKey : stale) {
             StatefulRedisConnection<K, V> connection = connections.get(connectionKey);
@@ -360,6 +365,7 @@ private Set<ConnectionKey> getStaleConnectionKeys() {
      */
     @Override
     public void setAutoFlushCommands(boolean autoFlush) {
+
         synchronized (stateLock) {
             this.autoFlushCommands = autoFlush;
         }
@@ -378,6 +384,7 @@ public void flushCommands() {
 
     @Override
     public void setReadFrom(ReadFrom readFrom) {
+
         synchronized (stateLock) {
             this.readFrom = readFrom;
             Arrays.fill(readers, null);
@@ -404,10 +411,10 @@ long getConnectionCount() {
      * Synchronize on {@code stateLock} to initiate a happens-before relation and clear the thread caches of other threads.
      */
     private void resetFastConnectionCache() {
+
         synchronized (stateLock) {
             Arrays.fill(writers, null);
             Arrays.fill(readers, null);
-
         }
     }
 
@@ -417,12 +424,15 @@ private static RuntimeException invalidConnectionPoint(String message) {
     }
 
     Supplier<SocketAddress> getSocketAddressSupplier(final ConnectionKey connectionKey) {
+
         return () -> {
+
             if (connectionKey.nodeId != null) {
                 SocketAddress socketAddress = getSocketAddress(connectionKey.nodeId);
                 logger.debug("Resolved SocketAddress {} using for Cluster node {}", socketAddress, connectionKey.nodeId);
                 return socketAddress;
             }
+
             SocketAddress socketAddress = new InetSocketAddress(connectionKey.host, connectionKey.port);
             logger.debug("Resolved SocketAddress {} using for Cluster node at {}:{}", socketAddress, connectionKey.host,
                     connectionKey.port);
@@ -431,6 +441,7 @@ Supplier<SocketAddress> getSocketAddressSupplier(final ConnectionKey connectionK
     }
 
     private SocketAddress getSocketAddress(String nodeId) {
+
         for (RedisClusterNode partition : partitions) {
             if (partition.getNodeId().equals(nodeId)) {
                 return SocketAddressResolver.resolve(partition.getUri(), redisClusterClient.getResources().dnsResolver());
@@ -465,6 +476,7 @@ public ConnectionKey(Intent intent, String host, int port) {
 
         @Override
         public boolean equals(Object o) {
+
             if (this == o)
                 return true;
             if (!(o instanceof ConnectionKey))
@@ -483,12 +495,26 @@ public boolean equals(Object o) {
 
         @Override
         public int hashCode() {
+
             int result = intent != null ? intent.name().hashCode() : 0;
             result = 31 * result + (nodeId != null ? nodeId.hashCode() : 0);
             result = 31 * result + (host != null ? host.hashCode() : 0);
             result = 31 * result + port;
             return result;
         }
+
+        @Override
+        public String toString() {
+
+            StringBuffer sb = new StringBuffer();
+            sb.append(getClass().getSimpleName());
+            sb.append(" [intent=").append(intent);
+            sb.append(", nodeId='").append(nodeId).append('\'');
+            sb.append(", host='").append(host).append('\'');
+            sb.append(", port=").append(port);
+            sb.append(']');
+            return sb.toString();
+        }
     }
 
     boolean validateClusterNodeMembership() {
@@ -588,6 +614,5 @@ public StatefulRedisConnection<K, V> apply(ConnectionKey key) {
     }
 
     interface ClusterNodeConnectionFactory<K, V> extends Function<ConnectionKey, StatefulRedisConnection<K, V>> {
-
     }
 }