class DynamicNodeSelection<API, CMD, K, V> extends AbstractNodeSelection<API, CMD, K, V> {

    private final ClusterDistributionChannelWriter writer;

    private final Predicate<RedisClusterNode> selector;

    private final ConnectionIntent connectionIntent;

    private final Function<StatefulRedisConnection<K, V>, API> apiExtractor;

    public DynamicNodeSelection(ClusterDistributionChannelWriter writer, Predicate<RedisClusterNode> selector,
                                ConnectionIntent connectionIntent, Function<StatefulRedisConnection<K, V>, API> apiExtractor) {

        this.selector = selector;
        this.connectionIntent = connectionIntent;
        this.writer = writer;
        this.apiExtractor = apiExtractor;
    }

    @Override
    protected CompletableFuture<StatefulRedisConnection<K, V>> getConnection(RedisClusterNode redisClusterNode) {<FILL_FUNCTION_BODY>}

    @Override
    protected CompletableFuture<API> getApi(RedisClusterNode redisClusterNode) {
        return getConnection(redisClusterNode).thenApply(apiExtractor);
    }

    @Override
    protected List<RedisClusterNode> nodes() {
        return writer.getPartitions().stream().filter(selector).collect(Collectors.toList());
    }

}


        RedisURI uri = redisClusterNode.getUri();
        AsyncClusterConnectionProvider async = (AsyncClusterConnectionProvider) writer.getClusterConnectionProvider();

        return async.getConnectionAsync(connectionIntent, uri.getHost(), uri.getPort());
    
<methods>public Map<io.lettuce.core.cluster.models.partitions.RedisClusterNode,API> asMap() ,public CMD commands() ,public API commands(int) ,public io.lettuce.core.cluster.models.partitions.RedisClusterNode node(int) ,public int size() <variables>