class MasterReplicaChannelWriter implements RedisChannelWriter {

    private MasterReplicaConnectionProvider<?, ?> masterReplicaConnectionProvider;

    private final ClientResources clientResources;

    private final ClientOptions clientOptions;

    private final io.lettuce.core.protocol.ReadOnlyCommands.ReadOnlyPredicate readOnlyCommands;

    private boolean closed = false;

    private boolean inTransaction;

    MasterReplicaChannelWriter(MasterReplicaConnectionProvider<?, ?> masterReplicaConnectionProvider,
            ClientResources clientResources, ClientOptions clientOptions) {
        this.masterReplicaConnectionProvider = masterReplicaConnectionProvider;
        this.clientResources = clientResources;
        this.clientOptions = clientOptions;
        this.readOnlyCommands = clientOptions.getReadOnlyCommands();
    }

    @Override
    @SuppressWarnings("unchecked")
    public <K, V, T> RedisCommand<K, V, T> write(RedisCommand<K, V, T> command) {

        LettuceAssert.notNull(command, "Command must not be null");

        if (closed) {
            throw new RedisException("Connection is closed");
        }

        if (isStartTransaction(command.getType())) {
            inTransaction = true;
        }

        ConnectionIntent connectionIntent = inTransaction ? ConnectionIntent.WRITE
                : (readOnlyCommands.isReadOnly(command) ? ConnectionIntent.READ : ConnectionIntent.WRITE);
        CompletableFuture<StatefulRedisConnection<K, V>> future = (CompletableFuture) masterReplicaConnectionProvider
                .getConnectionAsync(connectionIntent);

        if (isEndTransaction(command.getType())) {
            inTransaction = false;
        }

        if (isSuccessfullyCompleted(future)) {
            writeCommand(command, future.join(), null);
        } else {
            future.whenComplete((c, t) -> writeCommand(command, c, t));
        }

        return command;
    }

    @SuppressWarnings("unchecked")
    private static <K, V> void writeCommand(RedisCommand<K, V, ?> command, StatefulRedisConnection<K, V> connection,
            Throwable throwable) {

        if (throwable != null) {
            command.completeExceptionally(throwable);
            return;
        }

        try {
            connection.dispatch(command);
        } catch (Exception e) {
            command.completeExceptionally(e);
        }
    }

    @Override
    @SuppressWarnings("unchecked")
    public <K, V> Collection<RedisCommand<K, V, ?>> write(Collection<? extends RedisCommand<K, V, ?>> commands) {

        LettuceAssert.notNull(commands, "Commands must not be null");

        if (closed) {
            throw new RedisException("Connection is closed");
        }

        for (RedisCommand<K, V, ?> command : commands) {
            if (isStartTransaction(command.getType())) {
                inTransaction = true;
                break;
            }
        }

        // TODO: Retain order or retain ConnectionIntent preference?
        // Currently: Retain order
        ConnectionIntent connectionIntent = inTransaction ? ConnectionIntent.WRITE : getIntent(commands);

        CompletableFuture<StatefulRedisConnection<K, V>> future = (CompletableFuture) masterReplicaConnectionProvider
                .getConnectionAsync(connectionIntent);

        for (RedisCommand<K, V, ?> command : commands) {
            if (isEndTransaction(command.getType())) {
                inTransaction = false;
                break;
            }
        }

        if (isSuccessfullyCompleted(future)) {
            writeCommands(commands, future.join(), null);
        } else {
            future.whenComplete((c, t) -> writeCommands(commands, c, t));
        }

        return (Collection) commands;
    }

    @SuppressWarnings("unchecked")
    private static <K, V> void writeCommands(Collection<? extends RedisCommand<K, V, ?>> commands,
            StatefulRedisConnection<K, V> connection, Throwable throwable) {

        if (throwable != null) {
            commands.forEach(c -> c.completeExceptionally(throwable));
            return;
        }

        try {
            connection.dispatch(commands);
        } catch (Exception e) {
            commands.forEach(c -> c.completeExceptionally(e));
        }
    }

    /**
     * Optimization: Determine command intents and optimize for bulk execution preferring one node.
     * <p>
     * If there is only one ConnectionIntent, then we take the ConnectionIntent derived from the commands. If there is more than
     * one ConnectionIntent, then use {@link ConnectionIntent#WRITE}.
     *
     * @param commands {@link Collection} of {@link RedisCommand commands}.
     * @return the ConnectionIntent.
     */
    ConnectionIntent getIntent(Collection<? extends RedisCommand<?, ?, ?>> commands) {

        if (commands.isEmpty()) {
            return ConnectionIntent.WRITE;
        }

        for (RedisCommand<?, ?, ?> command : commands) {

            if (!readOnlyCommands.isReadOnly(command)) {
                return ConnectionIntent.WRITE;
            }
        }

        return ConnectionIntent.READ;
    }

    @Override
    public void close() {
        closeAsync().join();
    }

    @Override
    public CompletableFuture<Void> closeAsync() {<FILL_FUNCTION_BODY>}

    MasterReplicaConnectionProvider<?, ?> getUpstreamReplicaConnectionProvider() {
        return masterReplicaConnectionProvider;
    }

    @Override
    public void setConnectionFacade(ConnectionFacade connection) {
    }

    @Override
    public ClientResources getClientResources() {
        return clientResources;
    }

    @Override
    public void setAutoFlushCommands(boolean autoFlush) {
        masterReplicaConnectionProvider.setAutoFlushCommands(autoFlush);
    }

    @Override
    public void flushCommands() {
        masterReplicaConnectionProvider.flushCommands();
    }

    @Override
    public void reset() {
        masterReplicaConnectionProvider.reset();
    }

    /**
     * Set from which nodes data is read. The setting is used as default for read operations on this connection. See the
     * documentation for {@link ReadFrom} for more information.
     *
     * @param readFrom the read from setting, must not be {@code null}
     */
    public void setReadFrom(ReadFrom readFrom) {
        masterReplicaConnectionProvider.setReadFrom(readFrom);
    }

    /**
     * Gets the {@link ReadFrom} setting for this connection. Defaults to {@link ReadFrom#UPSTREAM} if not set.
     *
     * @return the read from setting
     */
    public ReadFrom getReadFrom() {
        return masterReplicaConnectionProvider.getReadFrom();
    }

    private static boolean isSuccessfullyCompleted(CompletableFuture<?> connectFuture) {
        return connectFuture.isDone() && !connectFuture.isCompletedExceptionally();
    }

    private static boolean isStartTransaction(ProtocolKeyword command) {
        return command.name().equals("MULTI");
    }

    private boolean isEndTransaction(ProtocolKeyword command) {
        return command.name().equals("EXEC") || command.name().equals("DISCARD");
    }

}


        if (closed) {
            return CompletableFuture.completedFuture(null);
        }

        closed = true;

        CompletableFuture<Void> future = null;

        if (masterReplicaConnectionProvider != null) {
            future = masterReplicaConnectionProvider.closeAsync();
            masterReplicaConnectionProvider = null;
        }

        if (future == null) {
            future = CompletableFuture.completedFuture(null);
        }

        return future;
    
<no_super_class>