class RedisClusterPubSubAsyncCommandsImpl<K, V> extends RedisPubSubAsyncCommandsImpl<K, V>
        implements RedisClusterPubSubAsyncCommands<K, V> {

    /**
     * Initialize a new connection.
     *
     * @param connection the connection .
     * @param codec Codec used to encode/decode keys and values.
     */
    public RedisClusterPubSubAsyncCommandsImpl(StatefulRedisPubSubConnection<K, V> connection, RedisCodec<K, V> codec) {
        super(connection, codec);
    }

    @Override
    public RedisFuture<Set<V>> georadius(K key, double longitude, double latitude, double distance, GeoArgs.Unit unit) {
        return super.georadius_ro(key, longitude, latitude, distance, unit);
    }

    @Override
    public RedisFuture<List<GeoWithin<V>>> georadius(K key, double longitude, double latitude, double distance,
            GeoArgs.Unit unit, GeoArgs geoArgs) {
        return super.georadius_ro(key, longitude, latitude, distance, unit, geoArgs);
    }

    @Override
    public RedisFuture<Set<V>> georadiusbymember(K key, V member, double distance, GeoArgs.Unit unit) {
        return super.georadiusbymember_ro(key, member, distance, unit);
    }

    @Override
    public RedisFuture<List<GeoWithin<V>>> georadiusbymember(K key, V member, double distance, GeoArgs.Unit unit,
            GeoArgs geoArgs) {
        return super.georadiusbymember_ro(key, member, distance, unit, geoArgs);
    }

    @Override
    public StatefulRedisClusterPubSubConnectionImpl<K, V> getStatefulConnection() {
        return (StatefulRedisClusterPubSubConnectionImpl<K, V>) super.getStatefulConnection();
    }

    @SuppressWarnings("unchecked")
    @Override
    public PubSubAsyncNodeSelection<K, V> nodes(Predicate<RedisClusterNode> predicate) {<FILL_FUNCTION_BODY>}

    private static class StaticPubSubAsyncNodeSelection<K, V>
            extends AbstractNodeSelection<RedisPubSubAsyncCommands<K, V>, NodeSelectionPubSubAsyncCommands<K, V>, K, V>
            implements PubSubAsyncNodeSelection<K, V> {

        private final List<RedisClusterNode> redisClusterNodes;

        private final ClusterDistributionChannelWriter writer;

        @SuppressWarnings("unchecked")
        public StaticPubSubAsyncNodeSelection(StatefulRedisClusterPubSubConnection<K, V> globalConnection,
                Predicate<RedisClusterNode> selector) {

            this.redisClusterNodes = globalConnection.getPartitions().stream().filter(selector).collect(Collectors.toList());
            writer = ((StatefulRedisClusterPubSubConnectionImpl) globalConnection).getClusterDistributionChannelWriter();
        }

        @Override
        protected CompletableFuture<RedisPubSubAsyncCommands<K, V>> getApi(RedisClusterNode redisClusterNode) {
            return getConnection(redisClusterNode).thenApply(StatefulRedisPubSubConnection::async);
        }

        protected List<RedisClusterNode> nodes() {
            return redisClusterNodes;
        }

        @SuppressWarnings("unchecked")
        protected CompletableFuture<StatefulRedisPubSubConnection<K, V>> getConnection(RedisClusterNode redisClusterNode) {

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

            return async.getConnectionAsync(ConnectionIntent.WRITE, uri.getHost(), uri.getPort())
                    .thenApply(it -> (StatefulRedisPubSubConnection<K, V>) it);
        }

    }

}


        PubSubAsyncNodeSelection<K, V> selection = new StaticPubSubAsyncNodeSelection<>(getStatefulConnection(), predicate);

        NodeSelectionInvocationHandler h = new NodeSelectionInvocationHandler((AbstractNodeSelection<?, ?, ?, ?>) selection,
                RedisPubSubAsyncCommands.class, ASYNC);
        return (PubSubAsyncNodeSelection<K, V>) Proxy.newProxyInstance(NodeSelectionSupport.class.getClassLoader(),
                new Class<?>[] { NodeSelectionPubSubAsyncCommands.class, PubSubAsyncNodeSelection.class }, h);
    
<methods>public void <init>(StatefulRedisPubSubConnection<K,V>, RedisCodec<K,V>) ,public StatefulRedisPubSubConnection<K,V> getStatefulConnection() ,public transient RedisFuture<java.lang.Void> psubscribe(K[]) ,public RedisFuture<java.lang.Long> publish(K, V) ,public RedisFuture<List<K>> pubsubChannels(K) ,public transient RedisFuture<Map<K,java.lang.Long>> pubsubNumsub(K[]) ,public RedisFuture<List<K>> pubsubShardChannels(K) ,public transient RedisFuture<Map<K,java.lang.Long>> pubsubShardNumsub(K[]) ,public transient RedisFuture<java.lang.Void> punsubscribe(K[]) ,public transient RedisFuture<java.lang.Void> ssubscribe(K[]) ,public transient RedisFuture<java.lang.Void> subscribe(K[]) ,public transient RedisFuture<java.lang.Void> unsubscribe(K[]) <variables>private final non-sealed PubSubCommandBuilder<K,V> commandBuilder