class CommandExpiryWriter implements RedisChannelWriter {

    private final RedisChannelWriter delegate;

    private final TimeoutSource source;

    private final TimeUnit timeUnit;

    private final ScheduledExecutorService executorService;

    private final Timer timer;

    private final boolean applyConnectionTimeout;

    private volatile long timeout = -1;

    /**
     * Create a new {@link CommandExpiryWriter}.
     *
     * @param delegate must not be {@code null}.
     * @param clientOptions must not be {@code null}.
     * @param clientResources must not be {@code null}.
     */
    public CommandExpiryWriter(RedisChannelWriter delegate, ClientOptions clientOptions, ClientResources clientResources) {

        LettuceAssert.notNull(delegate, "RedisChannelWriter must not be null");
        LettuceAssert.isTrue(isSupported(clientOptions), "Command timeout not enabled");
        LettuceAssert.notNull(clientResources, "ClientResources must not be null");

        TimeoutOptions timeoutOptions = clientOptions.getTimeoutOptions();
        this.delegate = delegate;
        this.source = timeoutOptions.getSource();
        this.applyConnectionTimeout = timeoutOptions.isApplyConnectionTimeout();
        this.timeUnit = source.getTimeUnit();
        this.executorService = clientResources.eventExecutorGroup();
        this.timer = clientResources.timer();
    }

    /**
     * Check whether {@link ClientOptions} is configured to timeout commands.
     *
     * @param clientOptions must not be {@code null}.
     * @return {@code true} if {@link ClientOptions} are configured to timeout commands.
     */
    public static boolean isSupported(ClientOptions clientOptions) {

        LettuceAssert.notNull(clientOptions, "ClientOptions must not be null");

        return isSupported(clientOptions.getTimeoutOptions());
    }

    private static boolean isSupported(TimeoutOptions timeoutOptions) {

        LettuceAssert.notNull(timeoutOptions, "TimeoutOptions must not be null");

        return timeoutOptions.isTimeoutCommands();
    }

    @Override
    public void setConnectionFacade(ConnectionFacade connectionFacade) {
        delegate.setConnectionFacade(connectionFacade);
    }

    @Override
    public ClientResources getClientResources() {
        return delegate.getClientResources();
    }

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

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

        potentiallyExpire(command, getExecutorService());
        return delegate.write(command);
    }

    @Override
    public <K, V> Collection<RedisCommand<K, V, ?>> write(Collection<? extends RedisCommand<K, V, ?>> redisCommands) {

        ScheduledExecutorService executorService = getExecutorService();

        for (RedisCommand<K, V, ?> command : redisCommands) {
            potentiallyExpire(command, executorService);
        }

        return delegate.write(redisCommands);
    }

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

    @Override
    public void close() {
        delegate.close();
    }

    @Override
    public CompletableFuture<Void> closeAsync() {
        return delegate.closeAsync();
    }

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

    public void setTimeout(Duration timeout) {
        this.timeout = timeUnit.convert(timeout.toNanos(), TimeUnit.NANOSECONDS);
    }

    public RedisChannelWriter getDelegate() {
        return delegate;
    }

    private ScheduledExecutorService getExecutorService() {
        return this.executorService;
    }

    private void potentiallyExpire(RedisCommand<?, ?, ?> command, ScheduledExecutorService executors) {<FILL_FUNCTION_BODY>}

}


        long timeout = applyConnectionTimeout ? this.timeout : source.getTimeout(command);

        if (timeout <= 0) {
            return;
        }

        Timeout commandTimeout = timer.newTimeout(t -> {
            if (!command.isDone()) {
                executors.submit(() -> command.completeExceptionally(
                        ExceptionFactory.createTimeoutException(Duration.ofNanos(timeUnit.toNanos(timeout)))));

            }
        }, timeout, timeUnit);

        if (command instanceof CompleteableCommand) {
            ((CompleteableCommand<?>) command).onComplete((o, o2) -> commandTimeout.cancel());
        }

    
<no_super_class>