class StaticPubSubReactiveNodeSelection<K, V>
            extends AbstractNodeSelection<RedisPubSubReactiveCommands<K, V>, NodeSelectionPubSubReactiveCommands<K, V>, K, V>
            implements PubSubReactiveNodeSelection<K, V> {

        private final List<RedisClusterNode> redisClusterNodes;

        private final ClusterDistributionChannelWriter writer;

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

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

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

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

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

    }

            RedisURI uri = redisClusterNode.getUri();

            AsyncClusterConnectionProvider async = (AsyncClusterConnectionProvider) writer.getClusterConnectionProvider();

            return async.getConnectionAsync(ConnectionIntent.WRITE, uri.getHost(), uri.getPort())
                    .thenApply(it -> (StatefulRedisPubSubConnection<K, V>) it);
        
<methods>public void <init>(StatefulRedisPubSubConnection<K,V>, RedisCodec<K,V>) ,public StatefulRedisPubSubConnection<K,V> getStatefulConnection() ,public Flux<ChannelMessage<K,V>> observeChannels() ,public Flux<ChannelMessage<K,V>> observeChannels(FluxSink.OverflowStrategy) ,public Flux<PatternMessage<K,V>> observePatterns() ,public Flux<PatternMessage<K,V>> observePatterns(FluxSink.OverflowStrategy) ,public transient Mono<java.lang.Void> psubscribe(K[]) ,public Mono<java.lang.Long> publish(K, V) ,public Flux<K> pubsubChannels(K) ,public transient Mono<Map<K,java.lang.Long>> pubsubNumsub(K[]) ,public Flux<K> pubsubShardChannels(K) ,public transient Mono<Map<K,java.lang.Long>> pubsubShardNumsub(K[]) ,public transient Mono<java.lang.Void> punsubscribe(K[]) ,public transient Mono<java.lang.Void> ssubscribe(K[]) ,public transient Mono<java.lang.Void> subscribe(K[]) ,public transient Mono<java.lang.Void> unsubscribe(K[]) <variables>private final non-sealed PubSubCommandBuilder<K,V> commandBuilder