class StatefulRedisClusterPubSubConnectionImpl<K, V> extends StatefulRedisPubSubConnectionImpl<K, V>
        implements StatefulRedisClusterPubSubConnection<K, V> {

    private final PubSubClusterEndpoint<K, V> endpoint;

    private final ClusterPushHandler clusterPushHandler;

    private volatile Partitions partitions;

    private volatile String nodeId;

    /**
     * Initialize a new connection.
     *
     * @param endpoint the channel writer.
     * @param clusterPushHandler the cluster push message handler.
     * @param codec Codec used to encode/decode keys and values.
     * @param timeout Maximum time to wait for a response.
     */
    public StatefulRedisClusterPubSubConnectionImpl(PubSubClusterEndpoint<K, V> endpoint, ClusterPushHandler clusterPushHandler,
            RedisChannelWriter writer, RedisCodec<K, V> codec, Duration timeout) {

        super(endpoint, writer, codec, timeout);

        this.endpoint = endpoint;
        this.clusterPushHandler = clusterPushHandler;
    }

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

    @Override
    protected RedisPubSubAsyncCommandsImpl<K, V> newRedisAsyncCommandsImpl() {
        return new RedisClusterPubSubAsyncCommandsImpl<>(this, codec);
    }

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

    @SuppressWarnings("unchecked")
    @Override
    protected RedisPubSubCommands<K, V> newRedisSyncCommandsImpl() {

        return (RedisPubSubCommands) Proxy.newProxyInstance(AbstractRedisClient.class.getClassLoader(),
                new Class<?>[] { RedisClusterPubSubCommands.class, RedisPubSubCommands.class }, syncInvocationHandler());
    }

    private InvocationHandler syncInvocationHandler() {
        return new ClusterFutureSyncInvocationHandler<K, V>(this, RedisPubSubAsyncCommands.class, PubSubNodeSelection.class,
                NodeSelectionPubSubCommands.class, async());
    }

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

    @Override
    protected RedisPubSubReactiveCommandsImpl<K, V> newRedisReactiveCommandsImpl() {
        return new RedisClusterPubSubReactiveCommandsImpl<K, V>(this, codec);
    }

    @Override
    protected List<RedisFuture<Void>> resubscribe() {

        async().clusterMyId().thenAccept(nodeId -> endpoint.setClusterNode(partitions.getPartitionByNodeId(nodeId)));

        return super.resubscribe();
    }

    @Override
    public StatefulRedisPubSubConnection<K, V> getConnection(String nodeId) {

        RedisURI redisURI = lookup(nodeId);

        if (redisURI == null) {
            throw new RedisException("NodeId " + nodeId + " does not belong to the cluster");
        }

        return (StatefulRedisPubSubConnection<K, V>) getClusterDistributionChannelWriter().getClusterConnectionProvider()
                .getConnection(ConnectionIntent.WRITE, nodeId);
    }

    @Override
    @SuppressWarnings("unchecked")
    public CompletableFuture<StatefulRedisPubSubConnection<K, V>> getConnectionAsync(String nodeId) {<FILL_FUNCTION_BODY>}

    @Override
    public StatefulRedisPubSubConnection<K, V> getConnection(String host, int port) {

        return (StatefulRedisPubSubConnection<K, V>) getClusterDistributionChannelWriter().getClusterConnectionProvider()
                .getConnection(ConnectionIntent.WRITE, host, port);
    }

    @Override
    public CompletableFuture<StatefulRedisPubSubConnection<K, V>> getConnectionAsync(String host, int port) {

        AsyncClusterConnectionProvider provider = (AsyncClusterConnectionProvider) getClusterDistributionChannelWriter()
                .getClusterConnectionProvider();

        return (CompletableFuture) provider.getConnectionAsync(ConnectionIntent.WRITE, host, port);
    }

    @Override
    public void activated() {

        if (!endpoint.isSubscribed()) {
            async.clusterMyId().thenAccept(this::setNodeId);
        }

        super.activated();
    }

    public void setPartitions(Partitions partitions) {

        LettuceAssert.notNull(partitions, "Partitions must not be null");

        this.partitions = partitions;

        String nodeId = getNodeId();
        if (nodeId != null && expireStaleConnections()) {

            if (partitions.getPartitionByNodeId(nodeId) == null) {
                endpoint.disconnect();
            }
        }

        getClusterDistributionChannelWriter().setPartitions(partitions);
    }

    private String getNodeId() {
        return this.nodeId;
    }

    private void setNodeId(String nodeId) {
        this.nodeId = nodeId;
    }

    public Partitions getPartitions() {
        return partitions;
    }

    @Override
    public void setNodeMessagePropagation(boolean enabled) {
        this.endpoint.setNodeMessagePropagation(enabled);
    }

    /**
     * Add a new {@link RedisClusterPubSubListener listener}.
     *
     * @param listener the listener, must not be {@code null}.
     */
    @Override
    public void addListener(RedisClusterPubSubListener<K, V> listener) {
        endpoint.addListener(listener);
    }

    /**
     * Remove an existing {@link RedisClusterPubSubListener listener}.
     *
     * @param listener the listener, must not be {@code null}.
     */
    @Override
    public void removeListener(RedisClusterPubSubListener<K, V> listener) {
        endpoint.removeListener(listener);
    }

    @Override
    public void addListener(RedisClusterPushListener listener) {
        clusterPushHandler.addListener(listener);
    }

    @Override
    public void removeListener(RedisClusterPushListener listener) {
        clusterPushHandler.removeListener(listener);
    }

    protected ClusterDistributionChannelWriter getClusterDistributionChannelWriter() {
        return (ClusterDistributionChannelWriter) super.getChannelWriter();
    }

    private RedisURI lookup(String nodeId) {

        for (RedisClusterNode partition : partitions) {
            if (partition.getNodeId().equals(nodeId)) {
                return partition.getUri();
            }
        }
        return null;
    }

    private boolean expireStaleConnections() {

        ClusterClientOptions options = getClusterClientOptions();
        return options == null || options.isCloseStaleConnections();
    }

    private ClusterClientOptions getClusterClientOptions() {

        ClientOptions options = getOptions();
        return options instanceof ClusterClientOptions ? (ClusterClientOptions) options : null;
    }

}


        RedisURI redisURI = lookup(nodeId);

        if (redisURI == null) {
            throw new RedisException("NodeId " + nodeId + " does not belong to the cluster");
        }

        AsyncClusterConnectionProvider provider = (AsyncClusterConnectionProvider) getClusterDistributionChannelWriter()
                .getClusterConnectionProvider();
        return (CompletableFuture) provider.getConnectionAsync(ConnectionIntent.WRITE, nodeId);

    
<methods>public void <init>(PubSubEndpoint<K,V>, io.lettuce.core.RedisChannelWriter, RedisCodec<K,V>, java.time.Duration) ,public void activated() ,public void addListener(RedisPubSubListener<K,V>) ,public RedisPubSubAsyncCommands<K,V> async() ,public RedisPubSubReactiveCommands<K,V> reactive() ,public void removeListener(RedisPubSubListener<K,V>) ,public RedisPubSubCommands<K,V> sync() <variables>private final non-sealed PubSubEndpoint<K,V> endpoint