class PubSubNodeConnectionFactory extends AbstractClusterNodeConnectionFactory<K, V> {

        PubSubNodeConnectionFactory(ClientResources clientResources) {
            super(clientResources);
        }

        @Override
        public ConnectionFuture<StatefulRedisConnection<K, V>> apply(ConnectionKey key) {<FILL_FUNCTION_BODY>}

    }


            if (key.nodeId != null) {

                // NodeId connections do not provide command recovery due to cluster reconfiguration
                return redisClusterClient.connectPubSubToNodeAsync((RedisCodec) redisCodec, key.nodeId,
                        getSocketAddressSupplier(key));
            }

            // Host and port connections do provide command recovery due to cluster reconfiguration
            return redisClusterClient.connectPubSubToNodeAsync((RedisCodec) redisCodec, key.host + ":" + key.port,
                    getSocketAddressSupplier(key));
        
<methods>public void <init>(io.lettuce.core.cluster.RedisClusterClient, io.lettuce.core.RedisChannelWriter, RedisCodec<K,V>, io.lettuce.core.cluster.ClusterEventListener) ,public void addListener(io.lettuce.core.cluster.api.push.RedisClusterPushListener) ,public void close() ,public CompletableFuture<java.lang.Void> closeAsync() ,public void closeStaleConnections() ,public void flushCommands() ,public StatefulRedisConnection<K,V> getConnection(io.lettuce.core.protocol.ConnectionIntent, int) ,public StatefulRedisConnection<K,V> getConnection(io.lettuce.core.protocol.ConnectionIntent, java.lang.String) ,public StatefulRedisConnection<K,V> getConnection(io.lettuce.core.protocol.ConnectionIntent, java.lang.String, int) ,public CompletableFuture<StatefulRedisConnection<K,V>> getConnectionAsync(io.lettuce.core.protocol.ConnectionIntent, int) ,public CompletableFuture<StatefulRedisConnection<K,V>> getConnectionAsync(io.lettuce.core.protocol.ConnectionIntent, java.lang.String) ,public CompletableFuture<StatefulRedisConnection<K,V>> getConnectionAsync(io.lettuce.core.protocol.ConnectionIntent, java.lang.String, int) ,public Collection<io.lettuce.core.cluster.api.push.RedisClusterPushListener> getPushListeners() ,public io.lettuce.core.ReadFrom getReadFrom() ,public void removeListener(io.lettuce.core.cluster.api.push.RedisClusterPushListener) ,public void reset() ,public void setAutoFlushCommands(boolean) ,public void setPartitions(io.lettuce.core.cluster.models.partitions.Partitions) ,public void setReadFrom(io.lettuce.core.ReadFrom) <variables>private boolean autoFlushCommands,private final non-sealed io.lettuce.core.cluster.ClusterEventListener clusterEventListener,private final non-sealed io.lettuce.core.RedisChannelWriter clusterWriter,private final non-sealed ClusterNodeConnectionFactory<K,V> connectionFactory,private final non-sealed AsyncConnectionProvider<io.lettuce.core.cluster.ClusterNodeConnectionFactory.ConnectionKey,StatefulRedisConnection<K,V>,ConnectionFuture<StatefulRedisConnection<K,V>>> connectionProvider,private final boolean debugEnabled,private static final InternalLogger logger,private final non-sealed io.lettuce.core.cluster.ClusterClientOptions options,private io.lettuce.core.cluster.models.partitions.Partitions partitions,private final List<io.lettuce.core.cluster.api.push.RedisClusterPushListener> pushListeners,private io.lettuce.core.ReadFrom readFrom,private final CompletableFuture<StatefulRedisConnection<K,V>>[][] readers,private final non-sealed io.lettuce.core.cluster.RedisClusterClient redisClusterClient,private final non-sealed RedisCodec<K,V> redisCodec,private final java.lang.Object stateLock,private final CompletableFuture<StatefulRedisConnection<K,V>>[] writers