class CommandListenerWriter implements RedisChannelWriter {

    private final RedisChannelWriter delegate;

    private final CommandListener listener;

    private final Clock clock = Clock.systemDefaultZone();

    public CommandListenerWriter(RedisChannelWriter delegate, List<CommandListener> listeners) {
        this.delegate = delegate;
        this.listener = new CommandListenerMulticaster(new ArrayList<>(listeners));
    }

    /**
     * Check whether the list of {@link CommandListener} is not empty.
     *
     * @param commandListeners must not be {@code null}.
     * @return {@code true} if the list of {@link CommandListener} is not empty.
     */
    public static boolean isSupported(List<CommandListener> commandListeners) {

        LettuceAssert.notNull(commandListeners, "CommandListeners must not be null");

        return !commandListeners.isEmpty();
    }

    @Override
    public <K, V, T> RedisCommand<K, V, T> write(RedisCommand<K, V, T> command) {
        long now = clock.millis();
        CommandStartedEvent startedEvent = new CommandStartedEvent((RedisCommand<Object, Object, Object>) command, now);
        listener.commandStarted(startedEvent);

        return delegate.write(new RedisCommandListenerCommand<>(command, clock, startedEvent.getContext(), now, listener));
    }

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

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

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

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

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

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

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

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

    public RedisChannelWriter getDelegate() {
        return this.delegate;
    }

    private static class RedisCommandListenerCommand<K, V, T> extends CommandWrapper<K, V, T> {

        private final Clock clock;

        private final Map<String, Object> context;

        private final long startedAt;

        private final CommandListener listener;

        public RedisCommandListenerCommand(RedisCommand<K, V, T> command, Clock clock, Map<String, Object> context,
                long startedAt, CommandListener listener) {
            super(command);

            this.clock = clock;
            this.context = context;
            this.startedAt = startedAt;
            this.listener = listener;
        }

        @Override
        protected void doOnComplete() {
            if (getOutput().hasError()) {

                CommandFailedEvent failedEvent = new CommandFailedEvent((RedisCommand<Object, Object, Object>) command, context,
                        ExceptionFactory.createExecutionException(getOutput().getError()));
                listener.commandFailed(failedEvent);
            } else {
                long now = clock.millis();
                CommandSucceededEvent succeededEvent = new CommandSucceededEvent((RedisCommand<Object, Object, Object>) command,
                        context, startedAt, now);
                listener.commandSucceeded(succeededEvent);
            }
        }

        @Override
        protected void doOnError(Throwable throwable) {
            CommandFailedEvent failedEvent = new CommandFailedEvent((RedisCommand<Object, Object, Object>) command, context,
                    throwable);
            listener.commandFailed(failedEvent);
        }

        @Override
        public void cancel() {
            super.cancel();
        }

    }

    /**
     * Wraps multiple command listeners into one multicaster.
     *
     * @author Mikhael Sokolov
     * @since 6.1
     */
    public static class CommandListenerMulticaster implements CommandListener {

        private final List<CommandListener> listeners;

        public CommandListenerMulticaster(List<CommandListener> listeners) {
            this.listeners = listeners;
        }

        @Override
        public void commandStarted(CommandStartedEvent event) {
            for (CommandListener listener : listeners) {
                listener.commandStarted(event);
            }
        }

        @Override
        public void commandSucceeded(CommandSucceededEvent event) {
            for (CommandListener listener : listeners) {
                listener.commandSucceeded(event);
            }
        }

        @Override
        public void commandFailed(CommandFailedEvent event) {
            for (CommandListener listener : listeners) {
                listener.commandFailed(event);
            }
        }

    }

}


        List<RedisCommandListenerCommand<K, V, ?>> listenedCommands = new ArrayList<>();
        long now = clock.millis();

        for (RedisCommand<K, V, ?> redisCommand : redisCommands) {

            CommandStartedEvent startedEvent = new CommandStartedEvent((RedisCommand<Object, Object, Object>) redisCommand,
                    now);
            listener.commandStarted(startedEvent);
            RedisCommandListenerCommand<K, V, ?> command = new RedisCommandListenerCommand<>(redisCommand, clock,
                    startedEvent.getContext(), now, listener);
            listenedCommands.add(command);
        }

        return delegate.write(listenedCommands);
    
<no_super_class>