class StatefulRedisPubSubConnectionImpl<K, V> extends StatefulRedisConnectionImpl<K, V>
        implements StatefulRedisPubSubConnection<K, V> {

    private final PubSubEndpoint<K, V> endpoint;

    /**
     * Initialize a new connection.
     *
     * @param endpoint the {@link PubSubEndpoint}
     * @param writer the writer used to write commands
     * @param codec Codec used to encode/decode keys and values.
     * @param timeout Maximum time to wait for a response.
     */
    public StatefulRedisPubSubConnectionImpl(PubSubEndpoint<K, V> endpoint, RedisChannelWriter writer, RedisCodec<K, V> codec,
            Duration timeout) {

        super(writer, endpoint, codec, timeout);
        this.endpoint = endpoint;
        endpoint.setConnectionState(getConnectionState());
    }

    /**
     * Add a new listener.
     *
     * @param listener Listener.
     */
    @Override
    public void addListener(RedisPubSubListener<K, V> listener) {
        endpoint.addListener(listener);
    }

    /**
     * Remove an existing listener.
     *
     * @param listener Listener.
     */
    @Override
    public void removeListener(RedisPubSubListener<K, V> listener) {
        endpoint.removeListener(listener);
    }

    @Override
    public RedisPubSubAsyncCommands<K, V> async() {
        return (RedisPubSubAsyncCommands<K, V>) async;
    }

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

    @Override
    public RedisPubSubCommands<K, V> sync() {
        return (RedisPubSubCommands<K, V>) sync;
    }

    @Override
    protected RedisPubSubCommands<K, V> newRedisSyncCommandsImpl() {
        return syncHandler(async(), RedisPubSubCommands.class);
    }

    @Override
    public RedisPubSubReactiveCommands<K, V> reactive() {
        return (RedisPubSubReactiveCommands<K, V>) reactive;
    }

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

    /**
     * Re-subscribe to all previously subscribed channels and patterns.
     *
     * @return list of the futures of the {@literal subscribe} and {@literal psubscribe} commands.
     */
    protected List<RedisFuture<Void>> resubscribe() {<FILL_FUNCTION_BODY>}

    @SuppressWarnings("unchecked")
    private <T> T[] toArray(Collection<T> c) {
        Class<T> cls = (Class<T>) c.iterator().next().getClass();
        T[] array = (T[]) Array.newInstance(cls, c.size());
        return c.toArray(array);
    }

    @Override
    public void activated() {
        super.activated();
        for (RedisFuture<Void> command : resubscribe()) {
            command.exceptionally(throwable -> {
                if (throwable instanceof RedisCommandExecutionException) {
                    InternalLoggerFactory.getInstance(getClass()).warn("Re-subscribe failed: " + command.getError());
                }
                return null;
            });
        }
    }

}


        List<RedisFuture<Void>> result = new ArrayList<>();

        if (endpoint.hasChannelSubscriptions()) {
            result.add(async().subscribe(toArray(endpoint.getChannels())));
        }

        if (endpoint.hasPatternSubscriptions()) {
            result.add(async().psubscribe(toArray(endpoint.getPatterns())));
        }

        return result;
    
<methods>public void <init>(io.lettuce.core.RedisChannelWriter, io.lettuce.core.protocol.PushHandler, RedisCodec<K,V>, java.time.Duration) ,public void addListener(io.lettuce.core.api.push.PushListener) ,public RedisAsyncCommands<K,V> async() ,public RedisCommand<K,V,T> dispatch(RedisCommand<K,V,T>) ,public Collection<RedisCommand<K,V,?>> dispatch(Collection<? extends RedisCommand<K,V,?>>) ,public RedisCodec<K,V> getCodec() ,public io.lettuce.core.ConnectionState getConnectionState() ,public boolean isMulti() ,public RedisReactiveCommands<K,V> reactive() ,public void removeListener(io.lettuce.core.api.push.PushListener) ,public void setClientName(java.lang.String) ,public RedisCommands<K,V> sync() <variables>protected final non-sealed RedisAsyncCommandsImpl<K,V> async,protected final non-sealed RedisCodec<K,V> codec,protected MultiOutput<K,V> multi,private final non-sealed io.lettuce.core.protocol.PushHandler pushHandler,protected final non-sealed RedisReactiveCommandsImpl<K,V> reactive,private final io.lettuce.core.ConnectionState state,protected final non-sealed RedisCommands<K,V> sync